Break up Database impl into multiple files, organized by topic

crates/collab/src/db/queries.rs

@@ -0,0 +1,10 @@
+use super::*;
+
+pub mod access_tokens;
+pub mod channels;
+pub mod contacts;
+pub mod projects;
+pub mod rooms;
+pub mod servers;
+pub mod signups;
+pub mod users;

crates/collab/src/db/queries/access_tokens.rs

@@ -0,0 +1,53 @@
+use super::*;
+
+impl Database {
+    pub async fn create_access_token(
+        &self,
+        user_id: UserId,
+        access_token_hash: &str,
+        max_access_token_count: usize,
+    ) -> Result<AccessTokenId> {
+        self.transaction(|tx| async {
+            let tx = tx;
+
+            let token = access_token::ActiveModel {
+                user_id: ActiveValue::set(user_id),
+                hash: ActiveValue::set(access_token_hash.into()),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+
+            access_token::Entity::delete_many()
+                .filter(
+                    access_token::Column::Id.in_subquery(
+                        Query::select()
+                            .column(access_token::Column::Id)
+                            .from(access_token::Entity)
+                            .and_where(access_token::Column::UserId.eq(user_id))
+                            .order_by(access_token::Column::Id, sea_orm::Order::Desc)
+                            .limit(10000)
+                            .offset(max_access_token_count as u64)
+                            .to_owned(),
+                    ),
+                )
+                .exec(&*tx)
+                .await?;
+            Ok(token.id)
+        })
+        .await
+    }
+
+    pub async fn get_access_token(
+        &self,
+        access_token_id: AccessTokenId,
+    ) -> Result<access_token::Model> {
+        self.transaction(|tx| async move {
+            Ok(access_token::Entity::find_by_id(access_token_id)
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such access token"))?)
+        })
+        .await
+    }
+}

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

@@ -0,0 +1,697 @@
+use super::*;
+
+impl Database {
+    pub async fn create_root_channel(
+        &self,
+        name: &str,
+        live_kit_room: &str,
+        creator_id: UserId,
+    ) -> Result<ChannelId> {
+        self.create_channel(name, None, live_kit_room, creator_id)
+            .await
+    }
+
+    pub async fn create_channel(
+        &self,
+        name: &str,
+        parent: Option<ChannelId>,
+        live_kit_room: &str,
+        creator_id: UserId,
+    ) -> Result<ChannelId> {
+        let name = Self::sanitize_channel_name(name)?;
+        self.transaction(move |tx| async move {
+            if let Some(parent) = parent {
+                self.check_user_is_channel_admin(parent, creator_id, &*tx)
+                    .await?;
+            }
+
+            let channel = channel::ActiveModel {
+                name: ActiveValue::Set(name.to_string()),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+
+            let channel_paths_stmt;
+            if let Some(parent) = parent {
+                let sql = r#"
+                    INSERT INTO channel_paths
+                    (id_path, channel_id)
+                    SELECT
+                        id_path || $1 || '/', $2
+                    FROM
+                        channel_paths
+                    WHERE
+                        channel_id = $3
+                "#;
+                channel_paths_stmt = Statement::from_sql_and_values(
+                    self.pool.get_database_backend(),
+                    sql,
+                    [
+                        channel.id.to_proto().into(),
+                        channel.id.to_proto().into(),
+                        parent.to_proto().into(),
+                    ],
+                );
+                tx.execute(channel_paths_stmt).await?;
+            } else {
+                channel_path::Entity::insert(channel_path::ActiveModel {
+                    channel_id: ActiveValue::Set(channel.id),
+                    id_path: ActiveValue::Set(format!("/{}/", channel.id)),
+                })
+                .exec(&*tx)
+                .await?;
+            }
+
+            channel_member::ActiveModel {
+                channel_id: ActiveValue::Set(channel.id),
+                user_id: ActiveValue::Set(creator_id),
+                accepted: ActiveValue::Set(true),
+                admin: ActiveValue::Set(true),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+
+            room::ActiveModel {
+                channel_id: ActiveValue::Set(Some(channel.id)),
+                live_kit_room: ActiveValue::Set(live_kit_room.to_string()),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+
+            Ok(channel.id)
+        })
+        .await
+    }
+
+    pub async fn remove_channel(
+        &self,
+        channel_id: ChannelId,
+        user_id: UserId,
+    ) -> Result<(Vec<ChannelId>, Vec<UserId>)> {
+        self.transaction(move |tx| async move {
+            self.check_user_is_channel_admin(channel_id, user_id, &*tx)
+                .await?;
+
+            // Don't remove descendant channels that have additional parents.
+            let mut channels_to_remove = self.get_channel_descendants([channel_id], &*tx).await?;
+            {
+                let mut channels_to_keep = channel_path::Entity::find()
+                    .filter(
+                        channel_path::Column::ChannelId
+                            .is_in(
+                                channels_to_remove
+                                    .keys()
+                                    .copied()
+                                    .filter(|&id| id != channel_id),
+                            )
+                            .and(
+                                channel_path::Column::IdPath
+                                    .not_like(&format!("%/{}/%", channel_id)),
+                            ),
+                    )
+                    .stream(&*tx)
+                    .await?;
+                while let Some(row) = channels_to_keep.next().await {
+                    let row = row?;
+                    channels_to_remove.remove(&row.channel_id);
+                }
+            }
+
+            let channel_ancestors = self.get_channel_ancestors(channel_id, &*tx).await?;
+            let members_to_notify: Vec<UserId> = channel_member::Entity::find()
+                .filter(channel_member::Column::ChannelId.is_in(channel_ancestors))
+                .select_only()
+                .column(channel_member::Column::UserId)
+                .distinct()
+                .into_values::<_, QueryUserIds>()
+                .all(&*tx)
+                .await?;
+
+            channel::Entity::delete_many()
+                .filter(channel::Column::Id.is_in(channels_to_remove.keys().copied()))
+                .exec(&*tx)
+                .await?;
+
+            Ok((channels_to_remove.into_keys().collect(), members_to_notify))
+        })
+        .await
+    }
+
+    pub async fn invite_channel_member(
+        &self,
+        channel_id: ChannelId,
+        invitee_id: UserId,
+        inviter_id: UserId,
+        is_admin: bool,
+    ) -> Result<()> {
+        self.transaction(move |tx| async move {
+            self.check_user_is_channel_admin(channel_id, inviter_id, &*tx)
+                .await?;
+
+            channel_member::ActiveModel {
+                channel_id: ActiveValue::Set(channel_id),
+                user_id: ActiveValue::Set(invitee_id),
+                accepted: ActiveValue::Set(false),
+                admin: ActiveValue::Set(is_admin),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+
+            Ok(())
+        })
+        .await
+    }
+
+    fn sanitize_channel_name(name: &str) -> Result<&str> {
+        let new_name = name.trim().trim_start_matches('#');
+        if new_name == "" {
+            Err(anyhow!("channel name can't be blank"))?;
+        }
+        Ok(new_name)
+    }
+
+    pub async fn rename_channel(
+        &self,
+        channel_id: ChannelId,
+        user_id: UserId,
+        new_name: &str,
+    ) -> Result<String> {
+        self.transaction(move |tx| async move {
+            let new_name = Self::sanitize_channel_name(new_name)?.to_string();
+
+            self.check_user_is_channel_admin(channel_id, user_id, &*tx)
+                .await?;
+
+            channel::ActiveModel {
+                id: ActiveValue::Unchanged(channel_id),
+                name: ActiveValue::Set(new_name.clone()),
+                ..Default::default()
+            }
+            .update(&*tx)
+            .await?;
+
+            Ok(new_name)
+        })
+        .await
+    }
+
+    pub async fn respond_to_channel_invite(
+        &self,
+        channel_id: ChannelId,
+        user_id: UserId,
+        accept: bool,
+    ) -> Result<()> {
+        self.transaction(move |tx| async move {
+            let rows_affected = if accept {
+                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
+            } else {
+                channel_member::ActiveModel {
+                    channel_id: ActiveValue::Unchanged(channel_id),
+                    user_id: ActiveValue::Unchanged(user_id),
+                    ..Default::default()
+                }
+                .delete(&*tx)
+                .await?
+                .rows_affected
+            };
+
+            if rows_affected == 0 {
+                Err(anyhow!("no such invitation"))?;
+            }
+
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn remove_channel_member(
+        &self,
+        channel_id: ChannelId,
+        member_id: UserId,
+        remover_id: UserId,
+    ) -> Result<()> {
+        self.transaction(|tx| async move {
+            self.check_user_is_channel_admin(channel_id, remover_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(())
+        })
+        .await
+    }
+
+    pub async fn get_channel_invites_for_user(&self, user_id: UserId) -> Result<Vec<Channel>> {
+        self.transaction(|tx| async move {
+            let channel_invites = channel_member::Entity::find()
+                .filter(
+                    channel_member::Column::UserId
+                        .eq(user_id)
+                        .and(channel_member::Column::Accepted.eq(false)),
+                )
+                .all(&*tx)
+                .await?;
+
+            let channels = channel::Entity::find()
+                .filter(
+                    channel::Column::Id.is_in(
+                        channel_invites
+                            .into_iter()
+                            .map(|channel_member| channel_member.channel_id),
+                    ),
+                )
+                .all(&*tx)
+                .await?;
+
+            let channels = channels
+                .into_iter()
+                .map(|channel| Channel {
+                    id: channel.id,
+                    name: channel.name,
+                    parent_id: None,
+                })
+                .collect();
+
+            Ok(channels)
+        })
+        .await
+    }
+
+    pub async fn get_channels_for_user(&self, user_id: UserId) -> Result<ChannelsForUser> {
+        self.transaction(|tx| async move {
+            let tx = tx;
+
+            let channel_memberships = channel_member::Entity::find()
+                .filter(
+                    channel_member::Column::UserId
+                        .eq(user_id)
+                        .and(channel_member::Column::Accepted.eq(true)),
+                )
+                .all(&*tx)
+                .await?;
+
+            let parents_by_child_id = self
+                .get_channel_descendants(channel_memberships.iter().map(|m| m.channel_id), &*tx)
+                .await?;
+
+            let channels_with_admin_privileges = channel_memberships
+                .iter()
+                .filter_map(|membership| membership.admin.then_some(membership.channel_id))
+                .collect();
+
+            let mut channels = Vec::with_capacity(parents_by_child_id.len());
+            {
+                let mut rows = channel::Entity::find()
+                    .filter(channel::Column::Id.is_in(parents_by_child_id.keys().copied()))
+                    .stream(&*tx)
+                    .await?;
+                while let Some(row) = rows.next().await {
+                    let row = row?;
+                    channels.push(Channel {
+                        id: row.id,
+                        name: row.name,
+                        parent_id: parents_by_child_id.get(&row.id).copied().flatten(),
+                    });
+                }
+            }
+
+            #[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)
+                }
+            }
+
+            Ok(ChannelsForUser {
+                channels,
+                channel_participants,
+                channels_with_admin_privileges,
+            })
+        })
+        .await
+    }
+
+    pub async fn get_channel_members(&self, id: ChannelId) -> Result<Vec<UserId>> {
+        self.transaction(|tx| async move { self.get_channel_members_internal(id, &*tx).await })
+            .await
+    }
+
+    pub async fn set_channel_member_admin(
+        &self,
+        channel_id: ChannelId,
+        from: UserId,
+        for_user: UserId,
+        admin: bool,
+    ) -> Result<()> {
+        self.transaction(|tx| async move {
+            self.check_user_is_channel_admin(channel_id, from, &*tx)
+                .await?;
+
+            let result = channel_member::Entity::update_many()
+                .filter(
+                    channel_member::Column::ChannelId
+                        .eq(channel_id)
+                        .and(channel_member::Column::UserId.eq(for_user)),
+                )
+                .set(channel_member::ActiveModel {
+                    admin: ActiveValue::set(admin),
+                    ..Default::default()
+                })
+                .exec(&*tx)
+                .await?;
+
+            if result.rows_affected == 0 {
+                Err(anyhow!("no such member"))?;
+            }
+
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn get_channel_member_details(
+        &self,
+        channel_id: ChannelId,
+        user_id: UserId,
+    ) -> Result<Vec<proto::ChannelMember>> {
+        self.transaction(|tx| async move {
+            self.check_user_is_channel_admin(channel_id, user_id, &*tx)
+                .await?;
+
+            #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
+            enum QueryMemberDetails {
+                UserId,
+                Admin,
+                IsDirectMember,
+                Accepted,
+            }
+
+            let tx = tx;
+            let ancestor_ids = self.get_channel_ancestors(channel_id, &*tx).await?;
+            let mut stream = channel_member::Entity::find()
+                .distinct()
+                .filter(channel_member::Column::ChannelId.is_in(ancestor_ids.iter().copied()))
+                .select_only()
+                .column(channel_member::Column::UserId)
+                .column(channel_member::Column::Admin)
+                .column_as(
+                    channel_member::Column::ChannelId.eq(channel_id),
+                    QueryMemberDetails::IsDirectMember,
+                )
+                .column(channel_member::Column::Accepted)
+                .order_by_asc(channel_member::Column::UserId)
+                .into_values::<_, QueryMemberDetails>()
+                .stream(&*tx)
+                .await?;
+
+            let mut rows = Vec::<proto::ChannelMember>::new();
+            while let Some(row) = stream.next().await {
+                let (user_id, is_admin, is_direct_member, is_invite_accepted): (
+                    UserId,
+                    bool,
+                    bool,
+                    bool,
+                ) = row?;
+                let kind = match (is_direct_member, is_invite_accepted) {
+                    (true, true) => proto::channel_member::Kind::Member,
+                    (true, false) => proto::channel_member::Kind::Invitee,
+                    (false, true) => proto::channel_member::Kind::AncestorMember,
+                    (false, false) => continue,
+                };
+                let user_id = user_id.to_proto();
+                let kind = kind.into();
+                if let Some(last_row) = rows.last_mut() {
+                    if last_row.user_id == user_id {
+                        if is_direct_member {
+                            last_row.kind = kind;
+                            last_row.admin = is_admin;
+                        }
+                        continue;
+                    }
+                }
+                rows.push(proto::ChannelMember {
+                    user_id,
+                    kind,
+                    admin: is_admin,
+                });
+            }
+
+            Ok(rows)
+        })
+        .await
+    }
+
+    pub async fn get_channel_members_internal(
+        &self,
+        id: ChannelId,
+        tx: &DatabaseTransaction,
+    ) -> Result<Vec<UserId>> {
+        let ancestor_ids = self.get_channel_ancestors(id, tx).await?;
+        let user_ids = channel_member::Entity::find()
+            .distinct()
+            .filter(
+                channel_member::Column::ChannelId
+                    .is_in(ancestor_ids.iter().copied())
+                    .and(channel_member::Column::Accepted.eq(true)),
+            )
+            .select_only()
+            .column(channel_member::Column::UserId)
+            .into_values::<_, QueryUserIds>()
+            .all(&*tx)
+            .await?;
+        Ok(user_ids)
+    }
+
+    pub async fn check_user_is_channel_member(
+        &self,
+        channel_id: ChannelId,
+        user_id: UserId,
+        tx: &DatabaseTransaction,
+    ) -> Result<()> {
+        let channel_ids = self.get_channel_ancestors(channel_id, tx).await?;
+        channel_member::Entity::find()
+            .filter(
+                channel_member::Column::ChannelId
+                    .is_in(channel_ids)
+                    .and(channel_member::Column::UserId.eq(user_id)),
+            )
+            .one(&*tx)
+            .await?
+            .ok_or_else(|| anyhow!("user is not a channel member or channel does not exist"))?;
+        Ok(())
+    }
+
+    pub async fn check_user_is_channel_admin(
+        &self,
+        channel_id: ChannelId,
+        user_id: UserId,
+        tx: &DatabaseTransaction,
+    ) -> Result<()> {
+        let channel_ids = self.get_channel_ancestors(channel_id, tx).await?;
+        channel_member::Entity::find()
+            .filter(
+                channel_member::Column::ChannelId
+                    .is_in(channel_ids)
+                    .and(channel_member::Column::UserId.eq(user_id))
+                    .and(channel_member::Column::Admin.eq(true)),
+            )
+            .one(&*tx)
+            .await?
+            .ok_or_else(|| anyhow!("user is not a channel admin or channel does not exist"))?;
+        Ok(())
+    }
+
+    pub async fn get_channel_ancestors(
+        &self,
+        channel_id: ChannelId,
+        tx: &DatabaseTransaction,
+    ) -> Result<Vec<ChannelId>> {
+        let paths = channel_path::Entity::find()
+            .filter(channel_path::Column::ChannelId.eq(channel_id))
+            .all(tx)
+            .await?;
+        let mut channel_ids = Vec::new();
+        for path in paths {
+            for id in path.id_path.trim_matches('/').split('/') {
+                if let Ok(id) = id.parse() {
+                    let id = ChannelId::from_proto(id);
+                    if let Err(ix) = channel_ids.binary_search(&id) {
+                        channel_ids.insert(ix, id);
+                    }
+                }
+            }
+        }
+        Ok(channel_ids)
+    }
+
+    async fn get_channel_descendants(
+        &self,
+        channel_ids: impl IntoIterator<Item = ChannelId>,
+        tx: &DatabaseTransaction,
+    ) -> Result<HashMap<ChannelId, Option<ChannelId>>> {
+        let mut values = String::new();
+        for id in channel_ids {
+            if !values.is_empty() {
+                values.push_str(", ");
+            }
+            write!(&mut values, "({})", id).unwrap();
+        }
+
+        if values.is_empty() {
+            return Ok(HashMap::default());
+        }
+
+        let sql = format!(
+            r#"
+            SELECT
+                descendant_paths.*
+            FROM
+                channel_paths parent_paths, channel_paths descendant_paths
+            WHERE
+                parent_paths.channel_id IN ({values}) AND
+                descendant_paths.id_path LIKE (parent_paths.id_path || '%')
+        "#
+        );
+
+        let stmt = Statement::from_string(self.pool.get_database_backend(), sql);
+
+        let mut parents_by_child_id = HashMap::default();
+        let mut paths = channel_path::Entity::find()
+            .from_raw_sql(stmt)
+            .stream(tx)
+            .await?;
+
+        while let Some(path) = paths.next().await {
+            let path = path?;
+            let ids = path.id_path.trim_matches('/').split('/');
+            let mut parent_id = None;
+            for id in ids {
+                if let Ok(id) = id.parse() {
+                    let id = ChannelId::from_proto(id);
+                    if id == path.channel_id {
+                        break;
+                    }
+                    parent_id = Some(id);
+                }
+            }
+            parents_by_child_id.insert(path.channel_id, parent_id);
+        }
+
+        Ok(parents_by_child_id)
+    }
+
+    /// Returns the channel with the given ID and:
+    /// - true if the user is a member
+    /// - false if the user hasn't accepted the invitation yet
+    pub async fn get_channel(
+        &self,
+        channel_id: ChannelId,
+        user_id: UserId,
+    ) -> Result<Option<(Channel, bool)>> {
+        self.transaction(|tx| async move {
+            let tx = tx;
+
+            let channel = channel::Entity::find_by_id(channel_id).one(&*tx).await?;
+
+            if let Some(channel) = channel {
+                if self
+                    .check_user_is_channel_member(channel_id, user_id, &*tx)
+                    .await
+                    .is_err()
+                {
+                    return Ok(None);
+                }
+
+                let channel_membership = channel_member::Entity::find()
+                    .filter(
+                        channel_member::Column::ChannelId
+                            .eq(channel_id)
+                            .and(channel_member::Column::UserId.eq(user_id)),
+                    )
+                    .one(&*tx)
+                    .await?;
+
+                let is_accepted = channel_membership
+                    .map(|membership| membership.accepted)
+                    .unwrap_or(false);
+
+                Ok(Some((
+                    Channel {
+                        id: channel.id,
+                        name: channel.name,
+                        parent_id: None,
+                    },
+                    is_accepted,
+                )))
+            } else {
+                Ok(None)
+            }
+        })
+        .await
+    }
+
+    pub async fn room_id_for_channel(&self, channel_id: ChannelId) -> Result<RoomId> {
+        self.transaction(|tx| async move {
+            let tx = tx;
+            let room = channel::Model {
+                id: channel_id,
+                ..Default::default()
+            }
+            .find_related(room::Entity)
+            .one(&*tx)
+            .await?
+            .ok_or_else(|| anyhow!("invalid channel"))?;
+            Ok(room.id)
+        })
+        .await
+    }
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
+enum QueryUserIds {
+    UserId,
+}

crates/collab/src/db/queries/contacts.rs

@@ -0,0 +1,298 @@
+use super::*;
+
+impl Database {
+    pub async fn get_contacts(&self, user_id: UserId) -> Result<Vec<Contact>> {
+        #[derive(Debug, FromQueryResult)]
+        struct ContactWithUserBusyStatuses {
+            user_id_a: UserId,
+            user_id_b: UserId,
+            a_to_b: bool,
+            accepted: bool,
+            should_notify: bool,
+            user_a_busy: bool,
+            user_b_busy: bool,
+        }
+
+        self.transaction(|tx| async move {
+            let user_a_participant = Alias::new("user_a_participant");
+            let user_b_participant = Alias::new("user_b_participant");
+            let mut db_contacts = contact::Entity::find()
+                .column_as(
+                    Expr::tbl(user_a_participant.clone(), room_participant::Column::Id)
+                        .is_not_null(),
+                    "user_a_busy",
+                )
+                .column_as(
+                    Expr::tbl(user_b_participant.clone(), room_participant::Column::Id)
+                        .is_not_null(),
+                    "user_b_busy",
+                )
+                .filter(
+                    contact::Column::UserIdA
+                        .eq(user_id)
+                        .or(contact::Column::UserIdB.eq(user_id)),
+                )
+                .join_as(
+                    JoinType::LeftJoin,
+                    contact::Relation::UserARoomParticipant.def(),
+                    user_a_participant,
+                )
+                .join_as(
+                    JoinType::LeftJoin,
+                    contact::Relation::UserBRoomParticipant.def(),
+                    user_b_participant,
+                )
+                .into_model::<ContactWithUserBusyStatuses>()
+                .stream(&*tx)
+                .await?;
+
+            let mut contacts = Vec::new();
+            while let Some(db_contact) = db_contacts.next().await {
+                let db_contact = db_contact?;
+                if db_contact.user_id_a == user_id {
+                    if db_contact.accepted {
+                        contacts.push(Contact::Accepted {
+                            user_id: db_contact.user_id_b,
+                            should_notify: db_contact.should_notify && db_contact.a_to_b,
+                            busy: db_contact.user_b_busy,
+                        });
+                    } else if db_contact.a_to_b {
+                        contacts.push(Contact::Outgoing {
+                            user_id: db_contact.user_id_b,
+                        })
+                    } else {
+                        contacts.push(Contact::Incoming {
+                            user_id: db_contact.user_id_b,
+                            should_notify: db_contact.should_notify,
+                        });
+                    }
+                } else if db_contact.accepted {
+                    contacts.push(Contact::Accepted {
+                        user_id: db_contact.user_id_a,
+                        should_notify: db_contact.should_notify && !db_contact.a_to_b,
+                        busy: db_contact.user_a_busy,
+                    });
+                } else if db_contact.a_to_b {
+                    contacts.push(Contact::Incoming {
+                        user_id: db_contact.user_id_a,
+                        should_notify: db_contact.should_notify,
+                    });
+                } else {
+                    contacts.push(Contact::Outgoing {
+                        user_id: db_contact.user_id_a,
+                    });
+                }
+            }
+
+            contacts.sort_unstable_by_key(|contact| contact.user_id());
+
+            Ok(contacts)
+        })
+        .await
+    }
+
+    pub async fn is_user_busy(&self, user_id: UserId) -> Result<bool> {
+        self.transaction(|tx| async move {
+            let participant = room_participant::Entity::find()
+                .filter(room_participant::Column::UserId.eq(user_id))
+                .one(&*tx)
+                .await?;
+            Ok(participant.is_some())
+        })
+        .await
+    }
+
+    pub async fn has_contact(&self, user_id_1: UserId, user_id_2: UserId) -> Result<bool> {
+        self.transaction(|tx| async move {
+            let (id_a, id_b) = if user_id_1 < user_id_2 {
+                (user_id_1, user_id_2)
+            } else {
+                (user_id_2, user_id_1)
+            };
+
+            Ok(contact::Entity::find()
+                .filter(
+                    contact::Column::UserIdA
+                        .eq(id_a)
+                        .and(contact::Column::UserIdB.eq(id_b))
+                        .and(contact::Column::Accepted.eq(true)),
+                )
+                .one(&*tx)
+                .await?
+                .is_some())
+        })
+        .await
+    }
+
+    pub async fn send_contact_request(&self, sender_id: UserId, receiver_id: UserId) -> Result<()> {
+        self.transaction(|tx| async move {
+            let (id_a, id_b, a_to_b) = if sender_id < receiver_id {
+                (sender_id, receiver_id, true)
+            } else {
+                (receiver_id, sender_id, false)
+            };
+
+            let rows_affected = contact::Entity::insert(contact::ActiveModel {
+                user_id_a: ActiveValue::set(id_a),
+                user_id_b: ActiveValue::set(id_b),
+                a_to_b: ActiveValue::set(a_to_b),
+                accepted: ActiveValue::set(false),
+                should_notify: ActiveValue::set(true),
+                ..Default::default()
+            })
+            .on_conflict(
+                OnConflict::columns([contact::Column::UserIdA, contact::Column::UserIdB])
+                    .values([
+                        (contact::Column::Accepted, true.into()),
+                        (contact::Column::ShouldNotify, false.into()),
+                    ])
+                    .action_and_where(
+                        contact::Column::Accepted.eq(false).and(
+                            contact::Column::AToB
+                                .eq(a_to_b)
+                                .and(contact::Column::UserIdA.eq(id_b))
+                                .or(contact::Column::AToB
+                                    .ne(a_to_b)
+                                    .and(contact::Column::UserIdA.eq(id_a))),
+                        ),
+                    )
+                    .to_owned(),
+            )
+            .exec_without_returning(&*tx)
+            .await?;
+
+            if rows_affected == 1 {
+                Ok(())
+            } else {
+                Err(anyhow!("contact already requested"))?
+            }
+        })
+        .await
+    }
+
+    /// Returns a bool indicating whether the removed contact had originally accepted or not
+    ///
+    /// Deletes the contact identified by the requester and responder ids, and then returns
+    /// whether the deleted contact had originally accepted or was a pending contact request.
+    ///
+    /// # Arguments
+    ///
+    /// * `requester_id` - The user that initiates this request
+    /// * `responder_id` - The user that will be removed
+    pub async fn remove_contact(&self, requester_id: UserId, responder_id: UserId) -> Result<bool> {
+        self.transaction(|tx| async move {
+            let (id_a, id_b) = if responder_id < requester_id {
+                (responder_id, requester_id)
+            } else {
+                (requester_id, responder_id)
+            };
+
+            let contact = contact::Entity::find()
+                .filter(
+                    contact::Column::UserIdA
+                        .eq(id_a)
+                        .and(contact::Column::UserIdB.eq(id_b)),
+                )
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such contact"))?;
+
+            contact::Entity::delete_by_id(contact.id).exec(&*tx).await?;
+            Ok(contact.accepted)
+        })
+        .await
+    }
+
+    pub async fn dismiss_contact_notification(
+        &self,
+        user_id: UserId,
+        contact_user_id: UserId,
+    ) -> Result<()> {
+        self.transaction(|tx| async move {
+            let (id_a, id_b, a_to_b) = if user_id < contact_user_id {
+                (user_id, contact_user_id, true)
+            } else {
+                (contact_user_id, user_id, false)
+            };
+
+            let result = contact::Entity::update_many()
+                .set(contact::ActiveModel {
+                    should_notify: ActiveValue::set(false),
+                    ..Default::default()
+                })
+                .filter(
+                    contact::Column::UserIdA
+                        .eq(id_a)
+                        .and(contact::Column::UserIdB.eq(id_b))
+                        .and(
+                            contact::Column::AToB
+                                .eq(a_to_b)
+                                .and(contact::Column::Accepted.eq(true))
+                                .or(contact::Column::AToB
+                                    .ne(a_to_b)
+                                    .and(contact::Column::Accepted.eq(false))),
+                        ),
+                )
+                .exec(&*tx)
+                .await?;
+            if result.rows_affected == 0 {
+                Err(anyhow!("no such contact request"))?
+            } else {
+                Ok(())
+            }
+        })
+        .await
+    }
+
+    pub async fn respond_to_contact_request(
+        &self,
+        responder_id: UserId,
+        requester_id: UserId,
+        accept: bool,
+    ) -> Result<()> {
+        self.transaction(|tx| async move {
+            let (id_a, id_b, a_to_b) = if responder_id < requester_id {
+                (responder_id, requester_id, false)
+            } else {
+                (requester_id, responder_id, true)
+            };
+            let rows_affected = if accept {
+                let result = contact::Entity::update_many()
+                    .set(contact::ActiveModel {
+                        accepted: ActiveValue::set(true),
+                        should_notify: ActiveValue::set(true),
+                        ..Default::default()
+                    })
+                    .filter(
+                        contact::Column::UserIdA
+                            .eq(id_a)
+                            .and(contact::Column::UserIdB.eq(id_b))
+                            .and(contact::Column::AToB.eq(a_to_b)),
+                    )
+                    .exec(&*tx)
+                    .await?;
+                result.rows_affected
+            } else {
+                let result = contact::Entity::delete_many()
+                    .filter(
+                        contact::Column::UserIdA
+                            .eq(id_a)
+                            .and(contact::Column::UserIdB.eq(id_b))
+                            .and(contact::Column::AToB.eq(a_to_b))
+                            .and(contact::Column::Accepted.eq(false)),
+                    )
+                    .exec(&*tx)
+                    .await?;
+
+                result.rows_affected
+            };
+
+            if rows_affected == 1 {
+                Ok(())
+            } else {
+                Err(anyhow!("no such contact request"))?
+            }
+        })
+        .await
+    }
+}

crates/collab/src/db/queries/projects.rs

@@ -0,0 +1,926 @@
+use super::*;
+
+impl Database {
+    pub async fn project_count_excluding_admins(&self) -> Result<usize> {
+        #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
+        enum QueryAs {
+            Count,
+        }
+
+        self.transaction(|tx| async move {
+            Ok(project::Entity::find()
+                .select_only()
+                .column_as(project::Column::Id.count(), QueryAs::Count)
+                .inner_join(user::Entity)
+                .filter(user::Column::Admin.eq(false))
+                .into_values::<_, QueryAs>()
+                .one(&*tx)
+                .await?
+                .unwrap_or(0i64) as usize)
+        })
+        .await
+    }
+
+    pub async fn share_project(
+        &self,
+        room_id: RoomId,
+        connection: ConnectionId,
+        worktrees: &[proto::WorktreeMetadata],
+    ) -> Result<RoomGuard<(ProjectId, proto::Room)>> {
+        self.room_transaction(room_id, |tx| async move {
+            let participant = room_participant::Entity::find()
+                .filter(
+                    Condition::all()
+                        .add(
+                            room_participant::Column::AnsweringConnectionId
+                                .eq(connection.id as i32),
+                        )
+                        .add(
+                            room_participant::Column::AnsweringConnectionServerId
+                                .eq(connection.owner_id as i32),
+                        ),
+                )
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("could not find participant"))?;
+            if participant.room_id != room_id {
+                return Err(anyhow!("shared project on unexpected room"))?;
+            }
+
+            let project = project::ActiveModel {
+                room_id: ActiveValue::set(participant.room_id),
+                host_user_id: ActiveValue::set(participant.user_id),
+                host_connection_id: ActiveValue::set(Some(connection.id as i32)),
+                host_connection_server_id: ActiveValue::set(Some(ServerId(
+                    connection.owner_id as i32,
+                ))),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+
+            if !worktrees.is_empty() {
+                worktree::Entity::insert_many(worktrees.iter().map(|worktree| {
+                    worktree::ActiveModel {
+                        id: ActiveValue::set(worktree.id as i64),
+                        project_id: ActiveValue::set(project.id),
+                        abs_path: ActiveValue::set(worktree.abs_path.clone()),
+                        root_name: ActiveValue::set(worktree.root_name.clone()),
+                        visible: ActiveValue::set(worktree.visible),
+                        scan_id: ActiveValue::set(0),
+                        completed_scan_id: ActiveValue::set(0),
+                    }
+                }))
+                .exec(&*tx)
+                .await?;
+            }
+
+            project_collaborator::ActiveModel {
+                project_id: ActiveValue::set(project.id),
+                connection_id: ActiveValue::set(connection.id as i32),
+                connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
+                user_id: ActiveValue::set(participant.user_id),
+                replica_id: ActiveValue::set(ReplicaId(0)),
+                is_host: ActiveValue::set(true),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+
+            let room = self.get_room(room_id, &tx).await?;
+            Ok((project.id, room))
+        })
+        .await
+    }
+
+    pub async fn unshare_project(
+        &self,
+        project_id: ProjectId,
+        connection: ConnectionId,
+    ) -> Result<RoomGuard<(proto::Room, Vec<ConnectionId>)>> {
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            let guest_connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
+
+            let project = project::Entity::find_by_id(project_id)
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("project not found"))?;
+            if project.host_connection()? == connection {
+                project::Entity::delete(project.into_active_model())
+                    .exec(&*tx)
+                    .await?;
+                let room = self.get_room(room_id, &tx).await?;
+                Ok((room, guest_connection_ids))
+            } else {
+                Err(anyhow!("cannot unshare a project hosted by another user"))?
+            }
+        })
+        .await
+    }
+
+    pub async fn update_project(
+        &self,
+        project_id: ProjectId,
+        connection: ConnectionId,
+        worktrees: &[proto::WorktreeMetadata],
+    ) -> Result<RoomGuard<(proto::Room, Vec<ConnectionId>)>> {
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            let project = project::Entity::find_by_id(project_id)
+                .filter(
+                    Condition::all()
+                        .add(project::Column::HostConnectionId.eq(connection.id as i32))
+                        .add(
+                            project::Column::HostConnectionServerId.eq(connection.owner_id as i32),
+                        ),
+                )
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such project"))?;
+
+            self.update_project_worktrees(project.id, worktrees, &tx)
+                .await?;
+
+            let guest_connection_ids = self.project_guest_connection_ids(project.id, &tx).await?;
+            let room = self.get_room(project.room_id, &tx).await?;
+            Ok((room, guest_connection_ids))
+        })
+        .await
+    }
+
+    pub(in crate::db) async fn update_project_worktrees(
+        &self,
+        project_id: ProjectId,
+        worktrees: &[proto::WorktreeMetadata],
+        tx: &DatabaseTransaction,
+    ) -> Result<()> {
+        if !worktrees.is_empty() {
+            worktree::Entity::insert_many(worktrees.iter().map(|worktree| worktree::ActiveModel {
+                id: ActiveValue::set(worktree.id as i64),
+                project_id: ActiveValue::set(project_id),
+                abs_path: ActiveValue::set(worktree.abs_path.clone()),
+                root_name: ActiveValue::set(worktree.root_name.clone()),
+                visible: ActiveValue::set(worktree.visible),
+                scan_id: ActiveValue::set(0),
+                completed_scan_id: ActiveValue::set(0),
+            }))
+            .on_conflict(
+                OnConflict::columns([worktree::Column::ProjectId, worktree::Column::Id])
+                    .update_column(worktree::Column::RootName)
+                    .to_owned(),
+            )
+            .exec(&*tx)
+            .await?;
+        }
+
+        worktree::Entity::delete_many()
+            .filter(worktree::Column::ProjectId.eq(project_id).and(
+                worktree::Column::Id.is_not_in(worktrees.iter().map(|worktree| worktree.id as i64)),
+            ))
+            .exec(&*tx)
+            .await?;
+
+        Ok(())
+    }
+
+    pub async fn update_worktree(
+        &self,
+        update: &proto::UpdateWorktree,
+        connection: ConnectionId,
+    ) -> Result<RoomGuard<Vec<ConnectionId>>> {
+        let project_id = ProjectId::from_proto(update.project_id);
+        let worktree_id = update.worktree_id as i64;
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            // Ensure the update comes from the host.
+            let _project = project::Entity::find_by_id(project_id)
+                .filter(
+                    Condition::all()
+                        .add(project::Column::HostConnectionId.eq(connection.id as i32))
+                        .add(
+                            project::Column::HostConnectionServerId.eq(connection.owner_id as i32),
+                        ),
+                )
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such project"))?;
+
+            // Update metadata.
+            worktree::Entity::update(worktree::ActiveModel {
+                id: ActiveValue::set(worktree_id),
+                project_id: ActiveValue::set(project_id),
+                root_name: ActiveValue::set(update.root_name.clone()),
+                scan_id: ActiveValue::set(update.scan_id as i64),
+                completed_scan_id: if update.is_last_update {
+                    ActiveValue::set(update.scan_id as i64)
+                } else {
+                    ActiveValue::default()
+                },
+                abs_path: ActiveValue::set(update.abs_path.clone()),
+                ..Default::default()
+            })
+            .exec(&*tx)
+            .await?;
+
+            if !update.updated_entries.is_empty() {
+                worktree_entry::Entity::insert_many(update.updated_entries.iter().map(|entry| {
+                    let mtime = entry.mtime.clone().unwrap_or_default();
+                    worktree_entry::ActiveModel {
+                        project_id: ActiveValue::set(project_id),
+                        worktree_id: ActiveValue::set(worktree_id),
+                        id: ActiveValue::set(entry.id as i64),
+                        is_dir: ActiveValue::set(entry.is_dir),
+                        path: ActiveValue::set(entry.path.clone()),
+                        inode: ActiveValue::set(entry.inode as i64),
+                        mtime_seconds: ActiveValue::set(mtime.seconds as i64),
+                        mtime_nanos: ActiveValue::set(mtime.nanos as i32),
+                        is_symlink: ActiveValue::set(entry.is_symlink),
+                        is_ignored: ActiveValue::set(entry.is_ignored),
+                        is_external: ActiveValue::set(entry.is_external),
+                        git_status: ActiveValue::set(entry.git_status.map(|status| status as i64)),
+                        is_deleted: ActiveValue::set(false),
+                        scan_id: ActiveValue::set(update.scan_id as i64),
+                    }
+                }))
+                .on_conflict(
+                    OnConflict::columns([
+                        worktree_entry::Column::ProjectId,
+                        worktree_entry::Column::WorktreeId,
+                        worktree_entry::Column::Id,
+                    ])
+                    .update_columns([
+                        worktree_entry::Column::IsDir,
+                        worktree_entry::Column::Path,
+                        worktree_entry::Column::Inode,
+                        worktree_entry::Column::MtimeSeconds,
+                        worktree_entry::Column::MtimeNanos,
+                        worktree_entry::Column::IsSymlink,
+                        worktree_entry::Column::IsIgnored,
+                        worktree_entry::Column::GitStatus,
+                        worktree_entry::Column::ScanId,
+                    ])
+                    .to_owned(),
+                )
+                .exec(&*tx)
+                .await?;
+            }
+
+            if !update.removed_entries.is_empty() {
+                worktree_entry::Entity::update_many()
+                    .filter(
+                        worktree_entry::Column::ProjectId
+                            .eq(project_id)
+                            .and(worktree_entry::Column::WorktreeId.eq(worktree_id))
+                            .and(
+                                worktree_entry::Column::Id
+                                    .is_in(update.removed_entries.iter().map(|id| *id as i64)),
+                            ),
+                    )
+                    .set(worktree_entry::ActiveModel {
+                        is_deleted: ActiveValue::Set(true),
+                        scan_id: ActiveValue::Set(update.scan_id as i64),
+                        ..Default::default()
+                    })
+                    .exec(&*tx)
+                    .await?;
+            }
+
+            if !update.updated_repositories.is_empty() {
+                worktree_repository::Entity::insert_many(update.updated_repositories.iter().map(
+                    |repository| worktree_repository::ActiveModel {
+                        project_id: ActiveValue::set(project_id),
+                        worktree_id: ActiveValue::set(worktree_id),
+                        work_directory_id: ActiveValue::set(repository.work_directory_id as i64),
+                        scan_id: ActiveValue::set(update.scan_id as i64),
+                        branch: ActiveValue::set(repository.branch.clone()),
+                        is_deleted: ActiveValue::set(false),
+                    },
+                ))
+                .on_conflict(
+                    OnConflict::columns([
+                        worktree_repository::Column::ProjectId,
+                        worktree_repository::Column::WorktreeId,
+                        worktree_repository::Column::WorkDirectoryId,
+                    ])
+                    .update_columns([
+                        worktree_repository::Column::ScanId,
+                        worktree_repository::Column::Branch,
+                    ])
+                    .to_owned(),
+                )
+                .exec(&*tx)
+                .await?;
+            }
+
+            if !update.removed_repositories.is_empty() {
+                worktree_repository::Entity::update_many()
+                    .filter(
+                        worktree_repository::Column::ProjectId
+                            .eq(project_id)
+                            .and(worktree_repository::Column::WorktreeId.eq(worktree_id))
+                            .and(
+                                worktree_repository::Column::WorkDirectoryId
+                                    .is_in(update.removed_repositories.iter().map(|id| *id as i64)),
+                            ),
+                    )
+                    .set(worktree_repository::ActiveModel {
+                        is_deleted: ActiveValue::Set(true),
+                        scan_id: ActiveValue::Set(update.scan_id as i64),
+                        ..Default::default()
+                    })
+                    .exec(&*tx)
+                    .await?;
+            }
+
+            let connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
+            Ok(connection_ids)
+        })
+        .await
+    }
+
+    pub async fn update_diagnostic_summary(
+        &self,
+        update: &proto::UpdateDiagnosticSummary,
+        connection: ConnectionId,
+    ) -> Result<RoomGuard<Vec<ConnectionId>>> {
+        let project_id = ProjectId::from_proto(update.project_id);
+        let worktree_id = update.worktree_id as i64;
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            let summary = update
+                .summary
+                .as_ref()
+                .ok_or_else(|| anyhow!("invalid summary"))?;
+
+            // Ensure the update comes from the host.
+            let project = project::Entity::find_by_id(project_id)
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such project"))?;
+            if project.host_connection()? != connection {
+                return Err(anyhow!("can't update a project hosted by someone else"))?;
+            }
+
+            // Update summary.
+            worktree_diagnostic_summary::Entity::insert(worktree_diagnostic_summary::ActiveModel {
+                project_id: ActiveValue::set(project_id),
+                worktree_id: ActiveValue::set(worktree_id),
+                path: ActiveValue::set(summary.path.clone()),
+                language_server_id: ActiveValue::set(summary.language_server_id as i64),
+                error_count: ActiveValue::set(summary.error_count as i32),
+                warning_count: ActiveValue::set(summary.warning_count as i32),
+                ..Default::default()
+            })
+            .on_conflict(
+                OnConflict::columns([
+                    worktree_diagnostic_summary::Column::ProjectId,
+                    worktree_diagnostic_summary::Column::WorktreeId,
+                    worktree_diagnostic_summary::Column::Path,
+                ])
+                .update_columns([
+                    worktree_diagnostic_summary::Column::LanguageServerId,
+                    worktree_diagnostic_summary::Column::ErrorCount,
+                    worktree_diagnostic_summary::Column::WarningCount,
+                ])
+                .to_owned(),
+            )
+            .exec(&*tx)
+            .await?;
+
+            let connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
+            Ok(connection_ids)
+        })
+        .await
+    }
+
+    pub async fn start_language_server(
+        &self,
+        update: &proto::StartLanguageServer,
+        connection: ConnectionId,
+    ) -> Result<RoomGuard<Vec<ConnectionId>>> {
+        let project_id = ProjectId::from_proto(update.project_id);
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            let server = update
+                .server
+                .as_ref()
+                .ok_or_else(|| anyhow!("invalid language server"))?;
+
+            // Ensure the update comes from the host.
+            let project = project::Entity::find_by_id(project_id)
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such project"))?;
+            if project.host_connection()? != connection {
+                return Err(anyhow!("can't update a project hosted by someone else"))?;
+            }
+
+            // Add the newly-started language server.
+            language_server::Entity::insert(language_server::ActiveModel {
+                project_id: ActiveValue::set(project_id),
+                id: ActiveValue::set(server.id as i64),
+                name: ActiveValue::set(server.name.clone()),
+                ..Default::default()
+            })
+            .on_conflict(
+                OnConflict::columns([
+                    language_server::Column::ProjectId,
+                    language_server::Column::Id,
+                ])
+                .update_column(language_server::Column::Name)
+                .to_owned(),
+            )
+            .exec(&*tx)
+            .await?;
+
+            let connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
+            Ok(connection_ids)
+        })
+        .await
+    }
+
+    pub async fn update_worktree_settings(
+        &self,
+        update: &proto::UpdateWorktreeSettings,
+        connection: ConnectionId,
+    ) -> Result<RoomGuard<Vec<ConnectionId>>> {
+        let project_id = ProjectId::from_proto(update.project_id);
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            // Ensure the update comes from the host.
+            let project = project::Entity::find_by_id(project_id)
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such project"))?;
+            if project.host_connection()? != connection {
+                return Err(anyhow!("can't update a project hosted by someone else"))?;
+            }
+
+            if let Some(content) = &update.content {
+                worktree_settings_file::Entity::insert(worktree_settings_file::ActiveModel {
+                    project_id: ActiveValue::Set(project_id),
+                    worktree_id: ActiveValue::Set(update.worktree_id as i64),
+                    path: ActiveValue::Set(update.path.clone()),
+                    content: ActiveValue::Set(content.clone()),
+                })
+                .on_conflict(
+                    OnConflict::columns([
+                        worktree_settings_file::Column::ProjectId,
+                        worktree_settings_file::Column::WorktreeId,
+                        worktree_settings_file::Column::Path,
+                    ])
+                    .update_column(worktree_settings_file::Column::Content)
+                    .to_owned(),
+                )
+                .exec(&*tx)
+                .await?;
+            } else {
+                worktree_settings_file::Entity::delete(worktree_settings_file::ActiveModel {
+                    project_id: ActiveValue::Set(project_id),
+                    worktree_id: ActiveValue::Set(update.worktree_id as i64),
+                    path: ActiveValue::Set(update.path.clone()),
+                    ..Default::default()
+                })
+                .exec(&*tx)
+                .await?;
+            }
+
+            let connection_ids = self.project_guest_connection_ids(project_id, &tx).await?;
+            Ok(connection_ids)
+        })
+        .await
+    }
+
+    pub async fn join_project(
+        &self,
+        project_id: ProjectId,
+        connection: ConnectionId,
+    ) -> Result<RoomGuard<(Project, ReplicaId)>> {
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            let participant = room_participant::Entity::find()
+                .filter(
+                    Condition::all()
+                        .add(
+                            room_participant::Column::AnsweringConnectionId
+                                .eq(connection.id as i32),
+                        )
+                        .add(
+                            room_participant::Column::AnsweringConnectionServerId
+                                .eq(connection.owner_id as i32),
+                        ),
+                )
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("must join a room first"))?;
+
+            let project = project::Entity::find_by_id(project_id)
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such project"))?;
+            if project.room_id != participant.room_id {
+                return Err(anyhow!("no such project"))?;
+            }
+
+            let mut collaborators = project
+                .find_related(project_collaborator::Entity)
+                .all(&*tx)
+                .await?;
+            let replica_ids = collaborators
+                .iter()
+                .map(|c| c.replica_id)
+                .collect::<HashSet<_>>();
+            let mut replica_id = ReplicaId(1);
+            while replica_ids.contains(&replica_id) {
+                replica_id.0 += 1;
+            }
+            let new_collaborator = project_collaborator::ActiveModel {
+                project_id: ActiveValue::set(project_id),
+                connection_id: ActiveValue::set(connection.id as i32),
+                connection_server_id: ActiveValue::set(ServerId(connection.owner_id as i32)),
+                user_id: ActiveValue::set(participant.user_id),
+                replica_id: ActiveValue::set(replica_id),
+                is_host: ActiveValue::set(false),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+            collaborators.push(new_collaborator);
+
+            let db_worktrees = project.find_related(worktree::Entity).all(&*tx).await?;
+            let mut worktrees = db_worktrees
+                .into_iter()
+                .map(|db_worktree| {
+                    (
+                        db_worktree.id as u64,
+                        Worktree {
+                            id: db_worktree.id as u64,
+                            abs_path: db_worktree.abs_path,
+                            root_name: db_worktree.root_name,
+                            visible: db_worktree.visible,
+                            entries: Default::default(),
+                            repository_entries: Default::default(),
+                            diagnostic_summaries: Default::default(),
+                            settings_files: Default::default(),
+                            scan_id: db_worktree.scan_id as u64,
+                            completed_scan_id: db_worktree.completed_scan_id as u64,
+                        },
+                    )
+                })
+                .collect::<BTreeMap<_, _>>();
+
+            // Populate worktree entries.
+            {
+                let mut db_entries = worktree_entry::Entity::find()
+                    .filter(
+                        Condition::all()
+                            .add(worktree_entry::Column::ProjectId.eq(project_id))
+                            .add(worktree_entry::Column::IsDeleted.eq(false)),
+                    )
+                    .stream(&*tx)
+                    .await?;
+                while let Some(db_entry) = db_entries.next().await {
+                    let db_entry = db_entry?;
+                    if let Some(worktree) = worktrees.get_mut(&(db_entry.worktree_id as u64)) {
+                        worktree.entries.push(proto::Entry {
+                            id: db_entry.id as u64,
+                            is_dir: db_entry.is_dir,
+                            path: db_entry.path,
+                            inode: db_entry.inode as u64,
+                            mtime: Some(proto::Timestamp {
+                                seconds: db_entry.mtime_seconds as u64,
+                                nanos: db_entry.mtime_nanos as u32,
+                            }),
+                            is_symlink: db_entry.is_symlink,
+                            is_ignored: db_entry.is_ignored,
+                            is_external: db_entry.is_external,
+                            git_status: db_entry.git_status.map(|status| status as i32),
+                        });
+                    }
+                }
+            }
+
+            // Populate repository entries.
+            {
+                let mut db_repository_entries = worktree_repository::Entity::find()
+                    .filter(
+                        Condition::all()
+                            .add(worktree_repository::Column::ProjectId.eq(project_id))
+                            .add(worktree_repository::Column::IsDeleted.eq(false)),
+                    )
+                    .stream(&*tx)
+                    .await?;
+                while let Some(db_repository_entry) = db_repository_entries.next().await {
+                    let db_repository_entry = db_repository_entry?;
+                    if let Some(worktree) =
+                        worktrees.get_mut(&(db_repository_entry.worktree_id as u64))
+                    {
+                        worktree.repository_entries.insert(
+                            db_repository_entry.work_directory_id as u64,
+                            proto::RepositoryEntry {
+                                work_directory_id: db_repository_entry.work_directory_id as u64,
+                                branch: db_repository_entry.branch,
+                            },
+                        );
+                    }
+                }
+            }
+
+            // Populate worktree diagnostic summaries.
+            {
+                let mut db_summaries = worktree_diagnostic_summary::Entity::find()
+                    .filter(worktree_diagnostic_summary::Column::ProjectId.eq(project_id))
+                    .stream(&*tx)
+                    .await?;
+                while let Some(db_summary) = db_summaries.next().await {
+                    let db_summary = db_summary?;
+                    if let Some(worktree) = worktrees.get_mut(&(db_summary.worktree_id as u64)) {
+                        worktree
+                            .diagnostic_summaries
+                            .push(proto::DiagnosticSummary {
+                                path: db_summary.path,
+                                language_server_id: db_summary.language_server_id as u64,
+                                error_count: db_summary.error_count as u32,
+                                warning_count: db_summary.warning_count as u32,
+                            });
+                    }
+                }
+            }
+
+            // Populate worktree settings files
+            {
+                let mut db_settings_files = worktree_settings_file::Entity::find()
+                    .filter(worktree_settings_file::Column::ProjectId.eq(project_id))
+                    .stream(&*tx)
+                    .await?;
+                while let Some(db_settings_file) = db_settings_files.next().await {
+                    let db_settings_file = db_settings_file?;
+                    if let Some(worktree) =
+                        worktrees.get_mut(&(db_settings_file.worktree_id as u64))
+                    {
+                        worktree.settings_files.push(WorktreeSettingsFile {
+                            path: db_settings_file.path,
+                            content: db_settings_file.content,
+                        });
+                    }
+                }
+            }
+
+            // Populate language servers.
+            let language_servers = project
+                .find_related(language_server::Entity)
+                .all(&*tx)
+                .await?;
+
+            let project = Project {
+                collaborators: collaborators
+                    .into_iter()
+                    .map(|collaborator| ProjectCollaborator {
+                        connection_id: collaborator.connection(),
+                        user_id: collaborator.user_id,
+                        replica_id: collaborator.replica_id,
+                        is_host: collaborator.is_host,
+                    })
+                    .collect(),
+                worktrees,
+                language_servers: language_servers
+                    .into_iter()
+                    .map(|language_server| proto::LanguageServer {
+                        id: language_server.id as u64,
+                        name: language_server.name,
+                    })
+                    .collect(),
+            };
+            Ok((project, replica_id as ReplicaId))
+        })
+        .await
+    }
+
+    pub async fn leave_project(
+        &self,
+        project_id: ProjectId,
+        connection: ConnectionId,
+    ) -> Result<RoomGuard<(proto::Room, LeftProject)>> {
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            let result = project_collaborator::Entity::delete_many()
+                .filter(
+                    Condition::all()
+                        .add(project_collaborator::Column::ProjectId.eq(project_id))
+                        .add(project_collaborator::Column::ConnectionId.eq(connection.id as i32))
+                        .add(
+                            project_collaborator::Column::ConnectionServerId
+                                .eq(connection.owner_id as i32),
+                        ),
+                )
+                .exec(&*tx)
+                .await?;
+            if result.rows_affected == 0 {
+                Err(anyhow!("not a collaborator on this project"))?;
+            }
+
+            let project = project::Entity::find_by_id(project_id)
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("no such project"))?;
+            let collaborators = project
+                .find_related(project_collaborator::Entity)
+                .all(&*tx)
+                .await?;
+            let connection_ids = collaborators
+                .into_iter()
+                .map(|collaborator| collaborator.connection())
+                .collect();
+
+            follower::Entity::delete_many()
+                .filter(
+                    Condition::any()
+                        .add(
+                            Condition::all()
+                                .add(follower::Column::ProjectId.eq(project_id))
+                                .add(
+                                    follower::Column::LeaderConnectionServerId
+                                        .eq(connection.owner_id),
+                                )
+                                .add(follower::Column::LeaderConnectionId.eq(connection.id)),
+                        )
+                        .add(
+                            Condition::all()
+                                .add(follower::Column::ProjectId.eq(project_id))
+                                .add(
+                                    follower::Column::FollowerConnectionServerId
+                                        .eq(connection.owner_id),
+                                )
+                                .add(follower::Column::FollowerConnectionId.eq(connection.id)),
+                        ),
+                )
+                .exec(&*tx)
+                .await?;
+
+            let room = self.get_room(project.room_id, &tx).await?;
+            let left_project = LeftProject {
+                id: project_id,
+                host_user_id: project.host_user_id,
+                host_connection_id: project.host_connection()?,
+                connection_ids,
+            };
+            Ok((room, left_project))
+        })
+        .await
+    }
+
+    pub async fn project_collaborators(
+        &self,
+        project_id: ProjectId,
+        connection_id: ConnectionId,
+    ) -> Result<RoomGuard<Vec<ProjectCollaborator>>> {
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            let collaborators = project_collaborator::Entity::find()
+                .filter(project_collaborator::Column::ProjectId.eq(project_id))
+                .all(&*tx)
+                .await?
+                .into_iter()
+                .map(|collaborator| ProjectCollaborator {
+                    connection_id: collaborator.connection(),
+                    user_id: collaborator.user_id,
+                    replica_id: collaborator.replica_id,
+                    is_host: collaborator.is_host,
+                })
+                .collect::<Vec<_>>();
+
+            if collaborators
+                .iter()
+                .any(|collaborator| collaborator.connection_id == connection_id)
+            {
+                Ok(collaborators)
+            } else {
+                Err(anyhow!("no such project"))?
+            }
+        })
+        .await
+    }
+
+    pub async fn project_connection_ids(
+        &self,
+        project_id: ProjectId,
+        connection_id: ConnectionId,
+    ) -> Result<RoomGuard<HashSet<ConnectionId>>> {
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            let mut collaborators = project_collaborator::Entity::find()
+                .filter(project_collaborator::Column::ProjectId.eq(project_id))
+                .stream(&*tx)
+                .await?;
+
+            let mut connection_ids = HashSet::default();
+            while let Some(collaborator) = collaborators.next().await {
+                let collaborator = collaborator?;
+                connection_ids.insert(collaborator.connection());
+            }
+
+            if connection_ids.contains(&connection_id) {
+                Ok(connection_ids)
+            } else {
+                Err(anyhow!("no such project"))?
+            }
+        })
+        .await
+    }
+
+    async fn project_guest_connection_ids(
+        &self,
+        project_id: ProjectId,
+        tx: &DatabaseTransaction,
+    ) -> Result<Vec<ConnectionId>> {
+        let mut collaborators = project_collaborator::Entity::find()
+            .filter(
+                project_collaborator::Column::ProjectId
+                    .eq(project_id)
+                    .and(project_collaborator::Column::IsHost.eq(false)),
+            )
+            .stream(tx)
+            .await?;
+
+        let mut guest_connection_ids = Vec::new();
+        while let Some(collaborator) = collaborators.next().await {
+            let collaborator = collaborator?;
+            guest_connection_ids.push(collaborator.connection());
+        }
+        Ok(guest_connection_ids)
+    }
+
+    pub async fn room_id_for_project(&self, project_id: ProjectId) -> Result<RoomId> {
+        self.transaction(|tx| async move {
+            let project = project::Entity::find_by_id(project_id)
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("project {} not found", project_id))?;
+            Ok(project.room_id)
+        })
+        .await
+    }
+
+    pub async fn follow(
+        &self,
+        project_id: ProjectId,
+        leader_connection: ConnectionId,
+        follower_connection: ConnectionId,
+    ) -> Result<RoomGuard<proto::Room>> {
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            follower::ActiveModel {
+                room_id: ActiveValue::set(room_id),
+                project_id: ActiveValue::set(project_id),
+                leader_connection_server_id: ActiveValue::set(ServerId(
+                    leader_connection.owner_id as i32,
+                )),
+                leader_connection_id: ActiveValue::set(leader_connection.id as i32),
+                follower_connection_server_id: ActiveValue::set(ServerId(
+                    follower_connection.owner_id as i32,
+                )),
+                follower_connection_id: ActiveValue::set(follower_connection.id as i32),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+
+            let room = self.get_room(room_id, &*tx).await?;
+            Ok(room)
+        })
+        .await
+    }
+
+    pub async fn unfollow(
+        &self,
+        project_id: ProjectId,
+        leader_connection: ConnectionId,
+        follower_connection: ConnectionId,
+    ) -> Result<RoomGuard<proto::Room>> {
+        let room_id = self.room_id_for_project(project_id).await?;
+        self.room_transaction(room_id, |tx| async move {
+            follower::Entity::delete_many()
+                .filter(
+                    Condition::all()
+                        .add(follower::Column::ProjectId.eq(project_id))
+                        .add(
+                            follower::Column::LeaderConnectionServerId
+                                .eq(leader_connection.owner_id),
+                        )
+                        .add(follower::Column::LeaderConnectionId.eq(leader_connection.id))
+                        .add(
+                            follower::Column::FollowerConnectionServerId
+                                .eq(follower_connection.owner_id),
+                        )
+                        .add(follower::Column::FollowerConnectionId.eq(follower_connection.id)),
+                )
+                .exec(&*tx)
+                .await?;
+
+            let room = self.get_room(room_id, &*tx).await?;
+            Ok(room)
+        })
+        .await
+    }
+}

crates/collab/src/db/queries/servers.rs

@@ -0,0 +1,81 @@
+use super::*;
+
+impl Database {
+    pub async fn create_server(&self, environment: &str) -> Result<ServerId> {
+        self.transaction(|tx| async move {
+            let server = server::ActiveModel {
+                environment: ActiveValue::set(environment.into()),
+                ..Default::default()
+            }
+            .insert(&*tx)
+            .await?;
+            Ok(server.id)
+        })
+        .await
+    }
+
+    pub async fn stale_room_ids(
+        &self,
+        environment: &str,
+        new_server_id: ServerId,
+    ) -> Result<Vec<RoomId>> {
+        self.transaction(|tx| async move {
+            #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
+            enum QueryAs {
+                RoomId,
+            }
+
+            let stale_server_epochs = self
+                .stale_server_ids(environment, new_server_id, &tx)
+                .await?;
+            Ok(room_participant::Entity::find()
+                .select_only()
+                .column(room_participant::Column::RoomId)
+                .distinct()
+                .filter(
+                    room_participant::Column::AnsweringConnectionServerId
+                        .is_in(stale_server_epochs),
+                )
+                .into_values::<_, QueryAs>()
+                .all(&*tx)
+                .await?)
+        })
+        .await
+    }
+
+    pub async fn delete_stale_servers(
+        &self,
+        environment: &str,
+        new_server_id: ServerId,
+    ) -> Result<()> {
+        self.transaction(|tx| async move {
+            server::Entity::delete_many()
+                .filter(
+                    Condition::all()
+                        .add(server::Column::Environment.eq(environment))
+                        .add(server::Column::Id.ne(new_server_id)),
+                )
+                .exec(&*tx)
+                .await?;
+            Ok(())
+        })
+        .await
+    }
+
+    async fn stale_server_ids(
+        &self,
+        environment: &str,
+        new_server_id: ServerId,
+        tx: &DatabaseTransaction,
+    ) -> Result<Vec<ServerId>> {
+        let stale_servers = server::Entity::find()
+            .filter(
+                Condition::all()
+                    .add(server::Column::Environment.eq(environment))
+                    .add(server::Column::Id.ne(new_server_id)),
+            )
+            .all(&*tx)
+            .await?;
+        Ok(stale_servers.into_iter().map(|server| server.id).collect())
+    }
+}

crates/collab/src/db/queries/signups.rs

@@ -0,0 +1,349 @@
+use super::*;
+use hyper::StatusCode;
+
+impl Database {
+    pub async fn create_invite_from_code(
+        &self,
+        code: &str,
+        email_address: &str,
+        device_id: Option<&str>,
+        added_to_mailing_list: bool,
+    ) -> Result<Invite> {
+        self.transaction(|tx| async move {
+            let existing_user = user::Entity::find()
+                .filter(user::Column::EmailAddress.eq(email_address))
+                .one(&*tx)
+                .await?;
+
+            if existing_user.is_some() {
+                Err(anyhow!("email address is already in use"))?;
+            }
+
+            let inviting_user_with_invites = match user::Entity::find()
+                .filter(
+                    user::Column::InviteCode
+                        .eq(code)
+                        .and(user::Column::InviteCount.gt(0)),
+                )
+                .one(&*tx)
+                .await?
+            {
+                Some(inviting_user) => inviting_user,
+                None => {
+                    return Err(Error::Http(
+                        StatusCode::UNAUTHORIZED,
+                        "unable to find an invite code with invites remaining".to_string(),
+                    ))?
+                }
+            };
+            user::Entity::update_many()
+                .filter(
+                    user::Column::Id
+                        .eq(inviting_user_with_invites.id)
+                        .and(user::Column::InviteCount.gt(0)),
+                )
+                .col_expr(
+                    user::Column::InviteCount,
+                    Expr::col(user::Column::InviteCount).sub(1),
+                )
+                .exec(&*tx)
+                .await?;
+
+            let signup = signup::Entity::insert(signup::ActiveModel {
+                email_address: ActiveValue::set(email_address.into()),
+                email_confirmation_code: ActiveValue::set(random_email_confirmation_code()),
+                email_confirmation_sent: ActiveValue::set(false),
+                inviting_user_id: ActiveValue::set(Some(inviting_user_with_invites.id)),
+                platform_linux: ActiveValue::set(false),
+                platform_mac: ActiveValue::set(false),
+                platform_windows: ActiveValue::set(false),
+                platform_unknown: ActiveValue::set(true),
+                device_id: ActiveValue::set(device_id.map(|device_id| device_id.into())),
+                added_to_mailing_list: ActiveValue::set(added_to_mailing_list),
+                ..Default::default()
+            })
+            .on_conflict(
+                OnConflict::column(signup::Column::EmailAddress)
+                    .update_column(signup::Column::InvitingUserId)
+                    .to_owned(),
+            )
+            .exec_with_returning(&*tx)
+            .await?;
+
+            Ok(Invite {
+                email_address: signup.email_address,
+                email_confirmation_code: signup.email_confirmation_code,
+            })
+        })
+        .await
+    }
+
+    pub async fn create_user_from_invite(
+        &self,
+        invite: &Invite,
+        user: NewUserParams,
+    ) -> Result<Option<NewUserResult>> {
+        self.transaction(|tx| async {
+            let tx = tx;
+            let signup = signup::Entity::find()
+                .filter(
+                    signup::Column::EmailAddress
+                        .eq(invite.email_address.as_str())
+                        .and(
+                            signup::Column::EmailConfirmationCode
+                                .eq(invite.email_confirmation_code.as_str()),
+                        ),
+                )
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| Error::Http(StatusCode::NOT_FOUND, "no such invite".to_string()))?;
+
+            if signup.user_id.is_some() {
+                return Ok(None);
+            }
+
+            let user = user::Entity::insert(user::ActiveModel {
+                email_address: ActiveValue::set(Some(invite.email_address.clone())),
+                github_login: ActiveValue::set(user.github_login.clone()),
+                github_user_id: ActiveValue::set(Some(user.github_user_id)),
+                admin: ActiveValue::set(false),
+                invite_count: ActiveValue::set(user.invite_count),
+                invite_code: ActiveValue::set(Some(random_invite_code())),
+                metrics_id: ActiveValue::set(Uuid::new_v4()),
+                ..Default::default()
+            })
+            .on_conflict(
+                OnConflict::column(user::Column::GithubLogin)
+                    .update_columns([
+                        user::Column::EmailAddress,
+                        user::Column::GithubUserId,
+                        user::Column::Admin,
+                    ])
+                    .to_owned(),
+            )
+            .exec_with_returning(&*tx)
+            .await?;
+
+            let mut signup = signup.into_active_model();
+            signup.user_id = ActiveValue::set(Some(user.id));
+            let signup = signup.update(&*tx).await?;
+
+            if let Some(inviting_user_id) = signup.inviting_user_id {
+                let (user_id_a, user_id_b, a_to_b) = if inviting_user_id < user.id {
+                    (inviting_user_id, user.id, true)
+                } else {
+                    (user.id, inviting_user_id, false)
+                };
+
+                contact::Entity::insert(contact::ActiveModel {
+                    user_id_a: ActiveValue::set(user_id_a),
+                    user_id_b: ActiveValue::set(user_id_b),
+                    a_to_b: ActiveValue::set(a_to_b),
+                    should_notify: ActiveValue::set(true),
+                    accepted: ActiveValue::set(true),
+                    ..Default::default()
+                })
+                .on_conflict(OnConflict::new().do_nothing().to_owned())
+                .exec_without_returning(&*tx)
+                .await?;
+            }
+
+            Ok(Some(NewUserResult {
+                user_id: user.id,
+                metrics_id: user.metrics_id.to_string(),
+                inviting_user_id: signup.inviting_user_id,
+                signup_device_id: signup.device_id,
+            }))
+        })
+        .await
+    }
+
+    pub async fn set_invite_count_for_user(&self, id: UserId, count: i32) -> Result<()> {
+        self.transaction(|tx| async move {
+            if count > 0 {
+                user::Entity::update_many()
+                    .filter(
+                        user::Column::Id
+                            .eq(id)
+                            .and(user::Column::InviteCode.is_null()),
+                    )
+                    .set(user::ActiveModel {
+                        invite_code: ActiveValue::set(Some(random_invite_code())),
+                        ..Default::default()
+                    })
+                    .exec(&*tx)
+                    .await?;
+            }
+
+            user::Entity::update_many()
+                .filter(user::Column::Id.eq(id))
+                .set(user::ActiveModel {
+                    invite_count: ActiveValue::set(count),
+                    ..Default::default()
+                })
+                .exec(&*tx)
+                .await?;
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn get_invite_code_for_user(&self, id: UserId) -> Result<Option<(String, i32)>> {
+        self.transaction(|tx| async move {
+            match user::Entity::find_by_id(id).one(&*tx).await? {
+                Some(user) if user.invite_code.is_some() => {
+                    Ok(Some((user.invite_code.unwrap(), user.invite_count)))
+                }
+                _ => Ok(None),
+            }
+        })
+        .await
+    }
+
+    pub async fn get_user_for_invite_code(&self, code: &str) -> Result<User> {
+        self.transaction(|tx| async move {
+            user::Entity::find()
+                .filter(user::Column::InviteCode.eq(code))
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| {
+                    Error::Http(
+                        StatusCode::NOT_FOUND,
+                        "that invite code does not exist".to_string(),
+                    )
+                })
+        })
+        .await
+    }
+
+    pub async fn create_signup(&self, signup: &NewSignup) -> Result<()> {
+        self.transaction(|tx| async move {
+            signup::Entity::insert(signup::ActiveModel {
+                email_address: ActiveValue::set(signup.email_address.clone()),
+                email_confirmation_code: ActiveValue::set(random_email_confirmation_code()),
+                email_confirmation_sent: ActiveValue::set(false),
+                platform_mac: ActiveValue::set(signup.platform_mac),
+                platform_windows: ActiveValue::set(signup.platform_windows),
+                platform_linux: ActiveValue::set(signup.platform_linux),
+                platform_unknown: ActiveValue::set(false),
+                editor_features: ActiveValue::set(Some(signup.editor_features.clone())),
+                programming_languages: ActiveValue::set(Some(signup.programming_languages.clone())),
+                device_id: ActiveValue::set(signup.device_id.clone()),
+                added_to_mailing_list: ActiveValue::set(signup.added_to_mailing_list),
+                ..Default::default()
+            })
+            .on_conflict(
+                OnConflict::column(signup::Column::EmailAddress)
+                    .update_columns([
+                        signup::Column::PlatformMac,
+                        signup::Column::PlatformWindows,
+                        signup::Column::PlatformLinux,
+                        signup::Column::EditorFeatures,
+                        signup::Column::ProgrammingLanguages,
+                        signup::Column::DeviceId,
+                        signup::Column::AddedToMailingList,
+                    ])
+                    .to_owned(),
+            )
+            .exec(&*tx)
+            .await?;
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn get_signup(&self, email_address: &str) -> Result<signup::Model> {
+        self.transaction(|tx| async move {
+            let signup = signup::Entity::find()
+                .filter(signup::Column::EmailAddress.eq(email_address))
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| {
+                    anyhow!("signup with email address {} doesn't exist", email_address)
+                })?;
+
+            Ok(signup)
+        })
+        .await
+    }
+
+    pub async fn get_waitlist_summary(&self) -> Result<WaitlistSummary> {
+        self.transaction(|tx| async move {
+            let query = "
+                SELECT
+                    COUNT(*) as count,
+                    COALESCE(SUM(CASE WHEN platform_linux THEN 1 ELSE 0 END), 0) as linux_count,
+                    COALESCE(SUM(CASE WHEN platform_mac THEN 1 ELSE 0 END), 0) as mac_count,
+                    COALESCE(SUM(CASE WHEN platform_windows THEN 1 ELSE 0 END), 0) as windows_count,
+                    COALESCE(SUM(CASE WHEN platform_unknown THEN 1 ELSE 0 END), 0) as unknown_count
+                FROM (
+                    SELECT *
+                    FROM signups
+                    WHERE
+                        NOT email_confirmation_sent
+                ) AS unsent
+            ";
+            Ok(
+                WaitlistSummary::find_by_statement(Statement::from_sql_and_values(
+                    self.pool.get_database_backend(),
+                    query.into(),
+                    vec![],
+                ))
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("invalid result"))?,
+            )
+        })
+        .await
+    }
+
+    pub async fn record_sent_invites(&self, invites: &[Invite]) -> Result<()> {
+        let emails = invites
+            .iter()
+            .map(|s| s.email_address.as_str())
+            .collect::<Vec<_>>();
+        self.transaction(|tx| async {
+            let tx = tx;
+            signup::Entity::update_many()
+                .filter(signup::Column::EmailAddress.is_in(emails.iter().copied()))
+                .set(signup::ActiveModel {
+                    email_confirmation_sent: ActiveValue::set(true),
+                    ..Default::default()
+                })
+                .exec(&*tx)
+                .await?;
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn get_unsent_invites(&self, count: usize) -> Result<Vec<Invite>> {
+        self.transaction(|tx| async move {
+            Ok(signup::Entity::find()
+                .select_only()
+                .column(signup::Column::EmailAddress)
+                .column(signup::Column::EmailConfirmationCode)
+                .filter(
+                    signup::Column::EmailConfirmationSent.eq(false).and(
+                        signup::Column::PlatformMac
+                            .eq(true)
+                            .or(signup::Column::PlatformUnknown.eq(true)),
+                    ),
+                )
+                .order_by_asc(signup::Column::CreatedAt)
+                .limit(count as u64)
+                .into_model()
+                .all(&*tx)
+                .await?)
+        })
+        .await
+    }
+}
+
+fn random_invite_code() -> String {
+    nanoid::nanoid!(16)
+}
+
+fn random_email_confirmation_code() -> String {
+    nanoid::nanoid!(64)
+}

crates/collab/src/db/queries/users.rs

@@ -0,0 +1,243 @@
+use super::*;
+
+impl Database {
+    pub async fn create_user(
+        &self,
+        email_address: &str,
+        admin: bool,
+        params: NewUserParams,
+    ) -> Result<NewUserResult> {
+        self.transaction(|tx| async {
+            let tx = tx;
+            let user = user::Entity::insert(user::ActiveModel {
+                email_address: ActiveValue::set(Some(email_address.into())),
+                github_login: ActiveValue::set(params.github_login.clone()),
+                github_user_id: ActiveValue::set(Some(params.github_user_id)),
+                admin: ActiveValue::set(admin),
+                metrics_id: ActiveValue::set(Uuid::new_v4()),
+                ..Default::default()
+            })
+            .on_conflict(
+                OnConflict::column(user::Column::GithubLogin)
+                    .update_column(user::Column::GithubLogin)
+                    .to_owned(),
+            )
+            .exec_with_returning(&*tx)
+            .await?;
+
+            Ok(NewUserResult {
+                user_id: user.id,
+                metrics_id: user.metrics_id.to_string(),
+                signup_device_id: None,
+                inviting_user_id: None,
+            })
+        })
+        .await
+    }
+
+    pub async fn get_user_by_id(&self, id: UserId) -> Result<Option<user::Model>> {
+        self.transaction(|tx| async move { Ok(user::Entity::find_by_id(id).one(&*tx).await?) })
+            .await
+    }
+
+    pub async fn get_users_by_ids(&self, ids: Vec<UserId>) -> Result<Vec<user::Model>> {
+        self.transaction(|tx| async {
+            let tx = tx;
+            Ok(user::Entity::find()
+                .filter(user::Column::Id.is_in(ids.iter().copied()))
+                .all(&*tx)
+                .await?)
+        })
+        .await
+    }
+
+    pub async fn get_user_by_github_login(&self, github_login: &str) -> Result<Option<User>> {
+        self.transaction(|tx| async move {
+            Ok(user::Entity::find()
+                .filter(user::Column::GithubLogin.eq(github_login))
+                .one(&*tx)
+                .await?)
+        })
+        .await
+    }
+
+    pub async fn get_or_create_user_by_github_account(
+        &self,
+        github_login: &str,
+        github_user_id: Option<i32>,
+        github_email: Option<&str>,
+    ) -> Result<Option<User>> {
+        self.transaction(|tx| async move {
+            let tx = &*tx;
+            if let Some(github_user_id) = github_user_id {
+                if let Some(user_by_github_user_id) = user::Entity::find()
+                    .filter(user::Column::GithubUserId.eq(github_user_id))
+                    .one(tx)
+                    .await?
+                {
+                    let mut user_by_github_user_id = user_by_github_user_id.into_active_model();
+                    user_by_github_user_id.github_login = ActiveValue::set(github_login.into());
+                    Ok(Some(user_by_github_user_id.update(tx).await?))
+                } else if let Some(user_by_github_login) = user::Entity::find()
+                    .filter(user::Column::GithubLogin.eq(github_login))
+                    .one(tx)
+                    .await?
+                {
+                    let mut user_by_github_login = user_by_github_login.into_active_model();
+                    user_by_github_login.github_user_id = ActiveValue::set(Some(github_user_id));
+                    Ok(Some(user_by_github_login.update(tx).await?))
+                } else {
+                    let user = user::Entity::insert(user::ActiveModel {
+                        email_address: ActiveValue::set(github_email.map(|email| email.into())),
+                        github_login: ActiveValue::set(github_login.into()),
+                        github_user_id: ActiveValue::set(Some(github_user_id)),
+                        admin: ActiveValue::set(false),
+                        invite_count: ActiveValue::set(0),
+                        invite_code: ActiveValue::set(None),
+                        metrics_id: ActiveValue::set(Uuid::new_v4()),
+                        ..Default::default()
+                    })
+                    .exec_with_returning(&*tx)
+                    .await?;
+                    Ok(Some(user))
+                }
+            } else {
+                Ok(user::Entity::find()
+                    .filter(user::Column::GithubLogin.eq(github_login))
+                    .one(tx)
+                    .await?)
+            }
+        })
+        .await
+    }
+
+    pub async fn get_all_users(&self, page: u32, limit: u32) -> Result<Vec<User>> {
+        self.transaction(|tx| async move {
+            Ok(user::Entity::find()
+                .order_by_asc(user::Column::GithubLogin)
+                .limit(limit as u64)
+                .offset(page as u64 * limit as u64)
+                .all(&*tx)
+                .await?)
+        })
+        .await
+    }
+
+    pub async fn get_users_with_no_invites(
+        &self,
+        invited_by_another_user: bool,
+    ) -> Result<Vec<User>> {
+        self.transaction(|tx| async move {
+            Ok(user::Entity::find()
+                .filter(
+                    user::Column::InviteCount
+                        .eq(0)
+                        .and(if invited_by_another_user {
+                            user::Column::InviterId.is_not_null()
+                        } else {
+                            user::Column::InviterId.is_null()
+                        }),
+                )
+                .all(&*tx)
+                .await?)
+        })
+        .await
+    }
+
+    pub async fn get_user_metrics_id(&self, id: UserId) -> Result<String> {
+        #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
+        enum QueryAs {
+            MetricsId,
+        }
+
+        self.transaction(|tx| async move {
+            let metrics_id: Uuid = user::Entity::find_by_id(id)
+                .select_only()
+                .column(user::Column::MetricsId)
+                .into_values::<_, QueryAs>()
+                .one(&*tx)
+                .await?
+                .ok_or_else(|| anyhow!("could not find user"))?;
+            Ok(metrics_id.to_string())
+        })
+        .await
+    }
+
+    pub async fn set_user_is_admin(&self, id: UserId, is_admin: bool) -> Result<()> {
+        self.transaction(|tx| async move {
+            user::Entity::update_many()
+                .filter(user::Column::Id.eq(id))
+                .set(user::ActiveModel {
+                    admin: ActiveValue::set(is_admin),
+                    ..Default::default()
+                })
+                .exec(&*tx)
+                .await?;
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn set_user_connected_once(&self, id: UserId, connected_once: bool) -> Result<()> {
+        self.transaction(|tx| async move {
+            user::Entity::update_many()
+                .filter(user::Column::Id.eq(id))
+                .set(user::ActiveModel {
+                    connected_once: ActiveValue::set(connected_once),
+                    ..Default::default()
+                })
+                .exec(&*tx)
+                .await?;
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn destroy_user(&self, id: UserId) -> Result<()> {
+        self.transaction(|tx| async move {
+            access_token::Entity::delete_many()
+                .filter(access_token::Column::UserId.eq(id))
+                .exec(&*tx)
+                .await?;
+            user::Entity::delete_by_id(id).exec(&*tx).await?;
+            Ok(())
+        })
+        .await
+    }
+
+    pub async fn fuzzy_search_users(&self, name_query: &str, limit: u32) -> Result<Vec<User>> {
+        self.transaction(|tx| async {
+            let tx = tx;
+            let like_string = Self::fuzzy_like_string(name_query);
+            let query = "
+                SELECT users.*
+                FROM users
+                WHERE github_login ILIKE $1
+                ORDER BY github_login <-> $2
+                LIMIT $3
+            ";
+
+            Ok(user::Entity::find()
+                .from_raw_sql(Statement::from_sql_and_values(
+                    self.pool.get_database_backend(),
+                    query.into(),
+                    vec![like_string.into(), name_query.into(), limit.into()],
+                ))
+                .all(&*tx)
+                .await?)
+        })
+        .await
+    }
+
+    pub fn fuzzy_like_string(string: &str) -> String {
+        let mut result = String::with_capacity(string.len() * 2 + 1);
+        for c in string.chars() {
+            if c.is_alphanumeric() {
+                result.push('%');
+                result.push(c);
+            }
+        }
+        result.push('%');
+        result
+    }
+}