Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Support very large channel membership lists (#11939)

Browse source 
Fixes the channel membership dialogue for the zed channel by not
downloading all 111k people in one go.

Release Notes:

- N/A
This commit is contained in:
Conrad Irwin 2024-05-16 20:02:25 -06:00 • committed by GitHub
parent df3bd40c56
commit 57b5bff299
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 270 additions and 261 deletions
Download patch file Download diff file Expand all files Collapse all files

3
crates/collab/src/db.rs
View file

@ -509,8 +509,7 @@ pub type NotificationBatch = Vec<(UserId, proto::Notification)>;
pub struct CreatedChannelMessage {
pub message_id: MessageId,
pub participant_connection_ids: Vec<ConnectionId>,
pub channel_members: Vec<UserId>,
pub participant_connection_ids: HashSet<ConnectionId>,
pub notifications: NotificationBatch,
}

21
crates/collab/src/db/queries/buffers.rs
View file

@ -440,12 +440,7 @@ impl Database {
channel_id: ChannelId,
user: UserId,
operations: &[proto::Operation],
) -> Result<(
Vec<ConnectionId>,
Vec<UserId>,
i32,
Vec<proto::VectorClockEntry>,
)> {
) -> Result<(HashSet<ConnectionId>, i32, Vec<proto::VectorClockEntry>)> {
self.transaction(move |tx| async move {
let channel = self.get_channel_internal(channel_id, &tx).await?;
@ -479,7 +474,6 @@ impl Database {
.filter_map(|op| operation_to_storage(op, &buffer, serialization_version))
.collect::<Vec<_>>();
let mut channel_members;
let max_version;
if !operations.is_empty() {
@ -504,12 +498,6 @@ impl Database {
)
.await?;
channel_members = self.get_channel_participants(&channel, &tx).await?;
let collaborators = self
.get_channel_buffer_collaborators_internal(channel_id, &tx)
.await?;
channel_members.retain(|member| !collaborators.contains(member));
buffer_operation::Entity::insert_many(operations)
.on_conflict(
OnConflict::columns([
@ -524,11 +512,10 @@ impl Database {
.exec(&*tx)
.await?;
} else {
channel_members = Vec::new();
max_version = Vec::new();
}
let mut connections = Vec::new();
let mut connections = HashSet::default();
let mut rows = channel_buffer_collaborator::Entity::find()
.filter(
Condition::all()
@ -538,13 +525,13 @@ impl Database {
.await?;
while let Some(row) = rows.next().await {
let row = row?;
connections.push(ConnectionId {
connections.insert(ConnectionId {
id: row.connection_id as u32,
owner_id: row.connection_server_id.0 as u32,
});
}
Ok((connections, channel_members, buffer.epoch, max_version))
Ok((connections, buffer.epoch, max_version))
})
.await
}

114
crates/collab/src/db/queries/channels.rs
View file

@ -3,7 +3,7 @@ use rpc::{
proto::{channel_member::Kind, ChannelBufferVersion, VectorClockEntry},
ErrorCode, ErrorCodeExt,
};
use sea_orm::TryGetableMany;
use sea_orm::{DbBackend, TryGetableMany};
impl Database {
#[cfg(test)]
@ -700,77 +700,73 @@ impl Database {
pub async fn get_channel_participant_details(
&self,
channel_id: ChannelId,
filter: &str,
limit: u64,
user_id: UserId,
) -> Result<Vec<proto::ChannelMember>> {
let (role, members) = self
) -> Result<(Vec<proto::ChannelMember>, Vec<proto::User>)> {
let members = self
.transaction(move |tx| async move {
let channel = self.get_channel_internal(channel_id, &tx).await?;
let role = self
.check_user_is_channel_participant(&channel, user_id, &tx)
self.check_user_is_channel_participant(&channel, user_id, &tx)
.await?;
Ok((
role,
self.get_channel_participant_details_internal(&channel, &tx)
.await?,
))
let mut query = channel_member::Entity::find()
.find_also_related(user::Entity)
.filter(channel_member::Column::ChannelId.eq(channel.root_id()));
if cfg!(any(test, sqlite)) && self.pool.get_database_backend() == DbBackend::Sqlite {
query = query.filter(Expr::cust_with_values(
"UPPER(github_login) LIKE ?",
[Self::fuzzy_like_string(&filter.to_uppercase())],
))
} else {
query = query.filter(Expr::cust_with_values(
"github_login ILIKE $1",
[Self::fuzzy_like_string(filter)],
))
}
let members = query.order_by(
Expr::cust(
"not role = 'admin', not role = 'member', not role = 'guest', not accepted, github_login",
),
sea_orm::Order::Asc,
)
.limit(limit)
.all(&*tx)
.await?;
Ok(members)
})
.await?;
if role == ChannelRole::Admin {
Ok(members
.into_iter()
.map(|channel_member| proto::ChannelMember {
role: channel_member.role.into(),
user_id: channel_member.user_id.to_proto(),
kind: if channel_member.accepted {
let mut users: Vec<proto::User> = Vec::with_capacity(members.len());
let members = members
.into_iter()
.map(|(member, user)| {
if let Some(user) = user {
users.push(proto::User {
id: user.id.to_proto(),
avatar_url: format!(
"https://github.com/{}.png?size=128",
user.github_login
),
github_login: user.github_login,
})
}
proto::ChannelMember {
role: member.role.into(),
user_id: member.user_id.to_proto(),
kind: if member.accepted {
Kind::Member
} else {
Kind::Invitee
}
.into(),
})
.collect())
} else {
return Ok(members
.into_iter()
.filter_map(|member| {
if !member.accepted {
return None;
}
Some(proto::ChannelMember {
role: member.role.into(),
user_id: member.user_id.to_proto(),
kind: Kind::Member.into(),
})
})
.collect());
}
}
}
})
.collect();
async fn get_channel_participant_details_internal(
&self,
channel: &channel::Model,
tx: &DatabaseTransaction,
) -> Result<Vec<channel_member::Model>> {
Ok(channel_member::Entity::find()
.filter(channel_member::Column::ChannelId.eq(channel.root_id()))
.all(tx)
.await?)
}
/// Returns the participants in the given channel.
pub async fn get_channel_participants(
&self,
channel: &channel::Model,
tx: &DatabaseTransaction,
) -> Result<Vec<UserId>> {
let participants = self
.get_channel_participant_details_internal(channel, tx)
.await?;
Ok(participants
.into_iter()
.map(|member| member.user_id)
.collect())
Ok((members, users))
}
/// Returns whether the given user is an admin in the specified channel.

8
crates/collab/src/db/queries/messages.rs
View file

@ -251,7 +251,7 @@ impl Database {
.await?;
let mut is_participant = false;
let mut participant_connection_ids = Vec::new();
let mut participant_connection_ids = HashSet::default();
let mut participant_user_ids = Vec::new();
while let Some(row) = rows.next().await {
let row = row?;
@ -259,7 +259,7 @@ impl Database {
is_participant = true;
}
participant_user_ids.push(row.user_id);
participant_connection_ids.push(row.connection());
participant_connection_ids.insert(row.connection());
}
drop(rows);
@ -336,13 +336,9 @@ impl Database {
}
}
let mut channel_members = self.get_channel_participants(&channel, &tx).await?;
channel_members.retain(|member| !participant_user_ids.contains(member));
Ok(CreatedChannelMessage {
message_id,
participant_connection_ids,
channel_members,
notifications,
})
})

50
crates/collab/src/db/tests/channel_tests.rs
View file

@ -1,7 +1,7 @@
use crate::{
db::{
tests::{channel_tree, new_test_connection, new_test_user},
Channel, ChannelId, ChannelRole, Database, NewUserParams, RoomId,
Channel, ChannelId, ChannelRole, Database, NewUserParams, RoomId, UserId,
},
test_both_dbs,
};
@ -40,15 +40,15 @@ async fn test_channels(db: &Arc<Database>) {
.await
.unwrap();
let mut members = db
.transaction(|tx| async move {
let channel = db.get_channel_internal(replace_id, &tx).await?;
db.get_channel_participants(&channel, &tx).await
})
let (members, _) = db
.get_channel_participant_details(replace_id, "", 10, a_id)
.await
.unwrap();
members.sort();
assert_eq!(members, &[a_id, b_id]);
let ids = members
.into_iter()
.map(|m| UserId::from_proto(m.user_id))
.collect::<Vec<_>>();
assert_eq!(ids, &[a_id, b_id]);
let rust_id = db.create_root_channel("rust", a_id).await.unwrap();
let cargo_id = db.create_sub_channel("cargo", rust_id, a_id).await.unwrap();
@ -195,8 +195,8 @@ async fn test_channel_invites(db: &Arc<Database>) {
assert_eq!(user_3_invites, &[channel_1_1]);
let mut members = db
.get_channel_participant_details(channel_1_1, user_1)
let (mut members, _) = db
.get_channel_participant_details(channel_1_1, "", 100, user_1)
.await
.unwrap();
@ -231,8 +231,8 @@ async fn test_channel_invites(db: &Arc<Database>) {
.await
.unwrap();
let members = db
.get_channel_participant_details(channel_1_3, user_1)
let (members, _) = db
.get_channel_participant_details(channel_1_3, "", 100, user_1)
.await
.unwrap();
assert_eq!(
@ -243,16 +243,16 @@ async fn test_channel_invites(db: &Arc<Database>) {
kind: proto::channel_member::Kind::Member.into(),
role: proto::ChannelRole::Admin.into(),
},
proto::ChannelMember {
user_id: user_2.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
role: proto::ChannelRole::Member.into(),
},
proto::ChannelMember {
user_id: user_3.to_proto(),
kind: proto::channel_member::Kind::Invitee.into(),
role: proto::ChannelRole::Admin.into(),
},
proto::ChannelMember {
user_id: user_2.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
role: proto::ChannelRole::Member.into(),
},
]
);
}
@ -482,8 +482,8 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
.await
.unwrap();
let mut members = db
.get_channel_participant_details(public_channel_id, admin)
let (mut members, _) = db
.get_channel_participant_details(public_channel_id, "", 100, admin)
.await
.unwrap();
@ -557,8 +557,8 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
.await
.is_err());
let mut members = db
.get_channel_participant_details(public_channel_id, admin)
let (mut members, _) = db
.get_channel_participant_details(public_channel_id, "", 100, admin)
.await
.unwrap();
@ -594,8 +594,8 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
.unwrap();
// currently people invited to parent channels are not shown here
let mut members = db
.get_channel_participant_details(public_channel_id, admin)
let (mut members, _) = db
.get_channel_participant_details(public_channel_id, "", 100, admin)
.await
.unwrap();
@ -663,8 +663,8 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
.await
.unwrap();
let mut members = db
.get_channel_participant_details(public_channel_id, admin)
let (mut members, _) = db
.get_channel_participant_details(public_channel_id, "", 100, admin)
.await
.unwrap();

99
crates/collab/src/rpc.rs
View file

@ -3683,10 +3683,15 @@ async fn get_channel_members(
) -> Result<()> {
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let members = db
.get_channel_participant_details(channel_id, session.user_id())
let limit = if request.limit == 0 {
u16::MAX as u64
} else {
request.limit
};
let (members, users) = db
.get_channel_participant_details(channel_id, &request.query, limit, session.user_id())
.await?;
response.send(proto::GetChannelMembersResponse { members })?;
response.send(proto::GetChannelMembersResponse { members, users })?;
Ok(())
}
@ -3886,13 +3891,13 @@ async fn update_channel_buffer(
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let (collaborators, non_collaborators, epoch, version) = db
let (collaborators, epoch, version) = db
.update_channel_buffer(channel_id, session.user_id(), &request.operations)
.await?;
channel_buffer_updated(
session.connection_id,
collaborators,
collaborators.clone(),
&proto::UpdateChannelBuffer {
channel_id: channel_id.to_proto(),
operations: request.operations,
@ -3902,25 +3907,29 @@ async fn update_channel_buffer(
let pool = &*session.connection_pool().await;
broadcast(
None,
non_collaborators
.iter()
.flat_map(|user_id| pool.user_connection_ids(*user_id)),
|peer_id| {
session.peer.send(
peer_id,
proto::UpdateChannels {
latest_channel_buffer_versions: vec![proto::ChannelBufferVersion {
channel_id: channel_id.to_proto(),
epoch: epoch as u64,
version: version.clone(),
}],
..Default::default()
},
)
},
);
let non_collaborators =
pool.channel_connection_ids(channel_id)
.filter_map(|(connection_id, _)| {
if collaborators.contains(&connection_id) {
None
} else {
Some(connection_id)
}
});
broadcast(None, non_collaborators, |peer_id| {
session.peer.send(
peer_id,
proto::UpdateChannels {
latest_channel_buffer_versions: vec![proto::ChannelBufferVersion {
channel_id: channel_id.to_proto(),
epoch: epoch as u64,
version: version.clone(),
}],
..Default::default()
},
)
});
Ok(())
}
@ -4048,7 +4057,6 @@ async fn send_channel_message(
let CreatedChannelMessage {
message_id,
participant_connection_ids,
channel_members,
notifications,
} = session
.db()
@ -4079,7 +4087,7 @@ async fn send_channel_message(
};
broadcast(
Some(session.connection_id),
participant_connection_ids,
participant_connection_ids.clone(),
|connection| {
session.peer.send(
connection,
@ -4095,24 +4103,27 @@ async fn send_channel_message(
})?;
let pool = &*session.connection_pool().await;
broadcast(
None,
channel_members
.iter()
.flat_map(|user_id| pool.user_connection_ids(*user_id)),
|peer_id| {
session.peer.send(
peer_id,
proto::UpdateChannels {
latest_channel_message_ids: vec![proto::ChannelMessageId {
channel_id: channel_id.to_proto(),
message_id: message_id.to_proto(),
}],
..Default::default()
},
)
},
);
let non_participants =
pool.channel_connection_ids(channel_id)
.filter_map(|(connection_id, _)| {
if participant_connection_ids.contains(&connection_id) {
None
} else {
Some(connection_id)
}
});
broadcast(None, non_participants, |peer_id| {
session.peer.send(
peer_id,
proto::UpdateChannels {
latest_channel_message_ids: vec![proto::ChannelMessageId {
channel_id: channel_id.to_proto(),
message_id: message_id.to_proto(),
}],
..Default::default()
},
)
});
send_notifications(pool, &session.peer, notifications);
Ok(())

2
crates/collab/src/tests/channel_tests.rs
View file

@ -99,7 +99,7 @@ async fn test_core_channels(
.channel_store()
.update(cx_a, |store, cx| {
assert!(!store.has_pending_channel_invite(channel_a_id, client_b.user_id().unwrap()));
store.get_channel_member_details(channel_a_id, cx)
store.fuzzy_search_members(channel_a_id, "".to_string(), 10, cx)
})
.await
.unwrap();