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

Guest roles (#3140)

Browse source 
Release Notes:

- Added a "guest" role to channels, and made that the default when a new
user joins a public channel.
This commit is contained in:
Max Brunsfeld 2023-10-24 17:29:14 +02:00 • committed by GitHub
commit cc9e92857b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
33 changed files with 2320 additions and 1711 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -435,18 +435,103 @@ pub struct NewUserResult {
pub signup_device_id: Option<String>,
}
#[derive(Debug)]
pub struct MoveChannelResult {
pub participants_to_update: HashMap<UserId, ChannelsForUser>,
pub participants_to_remove: HashSet<UserId>,
pub moved_channels: HashSet<ChannelId>,
}
#[derive(Debug)]
pub struct RenameChannelResult {
pub channel: Channel,
pub participants_to_update: HashMap<UserId, Channel>,
}
#[derive(Debug)]
pub struct CreateChannelResult {
pub channel: Channel,
pub participants_to_update: Vec<(UserId, ChannelsForUser)>,
}
#[derive(Debug)]
pub struct SetChannelVisibilityResult {
pub participants_to_update: HashMap<UserId, ChannelsForUser>,
pub participants_to_remove: HashSet<UserId>,
pub channels_to_remove: Vec<ChannelId>,
}
#[derive(Debug)]
pub struct MembershipUpdated {
pub channel_id: ChannelId,
pub new_channels: ChannelsForUser,
pub removed_channels: Vec<ChannelId>,
}
#[derive(Debug)]
pub enum SetMemberRoleResult {
InviteUpdated(Channel),
MembershipUpdated(MembershipUpdated),
}
#[derive(Debug)]
pub struct InviteMemberResult {
pub channel: Channel,
pub notifications: NotificationBatch,
}
#[derive(Debug)]
pub struct RespondToChannelInvite {
pub membership_update: Option<MembershipUpdated>,
pub notifications: NotificationBatch,
}
#[derive(Debug)]
pub struct RemoveChannelMemberResult {
pub membership_update: MembershipUpdated,
pub notification_id: Option<NotificationId>,
}
#[derive(FromQueryResult, Debug, PartialEq, Eq, Hash)]
pub struct Channel {
pub id: ChannelId,
pub name: String,
pub visibility: ChannelVisibility,
pub role: ChannelRole,
}
impl Channel {
pub fn to_proto(&self) -> proto::Channel {
proto::Channel {
id: self.id.to_proto(),
name: self.name.clone(),
visibility: self.visibility.into(),
role: self.role.into(),
}
}
}
#[derive(Debug, PartialEq, Eq, Hash)]
pub struct ChannelMember {
pub role: ChannelRole,
pub user_id: UserId,
pub kind: proto::channel_member::Kind,
}
impl ChannelMember {
pub fn to_proto(&self) -> proto::ChannelMember {
proto::ChannelMember {
role: self.role.into(),
user_id: self.user_id.to_proto(),
kind: self.kind.into(),
}
}
}
#[derive(Debug, PartialEq)]
pub struct ChannelsForUser {
pub channels: ChannelGraph,
pub channel_participants: HashMap<ChannelId, Vec<UserId>>,
pub channels_with_admin_privileges: HashSet<ChannelId>,
pub unseen_buffer_changes: Vec<proto::UnseenChannelBufferChange>,
pub channel_messages: Vec<proto::UnseenChannelMessage>,
}

18
crates/collab/src/db/ids.rs
View file

@ -84,7 +84,7 @@ id_type!(FlagId);
id_type!(NotificationId);
id_type!(NotificationKindId);
#[derive(Eq, PartialEq, Copy, Clone, Debug, EnumIter, DeriveActiveEnum, Default)]
#[derive(Eq, PartialEq, Copy, Clone, Debug, EnumIter, DeriveActiveEnum, Default, Hash)]
#[sea_orm(rs_type = "String", db_type = "String(None)")]
pub enum ChannelRole {
#[sea_orm(string_value = "admin")]
@ -116,6 +116,22 @@ impl ChannelRole {
other
}
}
pub fn can_see_all_descendants(&self) -> bool {
use ChannelRole::*;
match self {
Admin | Member => true,
Guest | Banned => false,
}
}
pub fn can_only_see_public_descendants(&self) -> bool {
use ChannelRole::*;
match self {
Guest => true,
Admin | Member | Banned => false,
}
}
}
impl From<proto::ChannelRole> for ChannelRole {

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

@ -16,7 +16,7 @@ impl Database {
connection: ConnectionId,
) -> Result<proto::JoinChannelBufferResponse> {
self.transaction(|tx| async move {
self.check_user_is_channel_member(channel_id, user_id, &tx)
self.check_user_is_channel_participant(channel_id, user_id, &tx)
.await?;
let buffer = channel::Model {
@ -131,7 +131,7 @@ impl Database {
for client_buffer in buffers {
let channel_id = ChannelId::from_proto(client_buffer.channel_id);
if self
.check_user_is_channel_member(channel_id, user_id, &*tx)
.check_user_is_channel_participant(channel_id, user_id, &*tx)
.await
.is_err()
{
@ -482,9 +482,7 @@ impl Database {
)
.await?;
channel_members = self
.get_channel_participants_internal(channel_id, &*tx)
.await?;
channel_members = self.get_channel_participants(channel_id, &*tx).await?;
let collaborators = self
.get_channel_buffer_collaborators_internal(channel_id, &*tx)
.await?;

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

File diff suppressed because it is too large Load diff

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

@ -12,7 +12,7 @@ impl Database {
user_id: UserId,
) -> Result<()> {
self.transaction(|tx| async move {
self.check_user_is_channel_member(channel_id, user_id, &*tx)
self.check_user_is_channel_participant(channel_id, user_id, &*tx)
.await?;
channel_chat_participant::ActiveModel {
id: ActiveValue::NotSet,
@ -80,7 +80,7 @@ impl Database {
before_message_id: Option<MessageId>,
) -> Result<Vec<proto::ChannelMessage>> {
self.transaction(|tx| async move {
self.check_user_is_channel_member(channel_id, user_id, &*tx)
self.check_user_is_channel_participant(channel_id, user_id, &*tx)
.await?;
let mut condition =
@ -203,6 +203,9 @@ impl Database {
nonce: u128,
) -> Result<CreatedChannelMessage> {
self.transaction(|tx| async move {
self.check_user_is_channel_participant(channel_id, user_id, &*tx)
.await?;
let mut rows = channel_chat_participant::Entity::find()
.filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
.stream(&*tx)
@ -307,9 +310,7 @@ impl Database {
}
}
let mut channel_members = self
.get_channel_participants_internal(channel_id, &*tx)
.await?;
let mut channel_members = self.get_channel_participants(channel_id, &*tx).await?;
channel_members.retain(|member| !participant_user_ids.contains(member));
Ok(CreatedChannelMessage {

14
crates/collab/src/db/queries/rooms.rs
View file

@ -53,9 +53,7 @@ impl Database {
let (channel_id, room) = self.get_channel_room(room_id, &tx).await?;
let channel_members;
if let Some(channel_id) = channel_id {
channel_members = self
.get_channel_participants_internal(channel_id, &tx)
.await?;
channel_members = self.get_channel_participants(channel_id, &tx).await?;
} else {
channel_members = Vec::new();
@ -423,9 +421,7 @@ impl Database {
.await?;
let room = self.get_room(room_id, &tx).await?;
let channel_members = self
.get_channel_participants_internal(channel_id, &tx)
.await?;
let channel_members = self.get_channel_participants(channel_id, &tx).await?;
Ok(JoinRoom {
room,
channel_id: Some(channel_id),
@ -724,8 +720,7 @@ impl Database {
let (channel_id, room) = self.get_channel_room(room_id, &tx).await?;
let channel_members = if let Some(channel_id) = channel_id {
self.get_channel_participants_internal(channel_id, &tx)
.await?
self.get_channel_participants(channel_id, &tx).await?
} else {
Vec::new()
};
@ -883,8 +878,7 @@ impl Database {
};
let channel_members = if let Some(channel_id) = channel_id {
self.get_channel_participants_internal(channel_id, &tx)
.await?
self.get_channel_participants(channel_id, &tx).await?
} else {
Vec::new()
};

18
crates/collab/src/db/tests.rs
View file

@ -11,7 +11,7 @@ use rpc::proto::ChannelEdge;
use sea_orm::ConnectionTrait;
use sqlx::migrate::MigrateDatabase;
use std::sync::{
atomic::{AtomicI32, Ordering::SeqCst},
atomic::{AtomicI32, AtomicU32, Ordering::SeqCst},
Arc,
};
@ -154,17 +154,21 @@ impl Drop for TestDb {
}
/// The second tuples are (channel_id, parent)
fn graph(channels: &[(ChannelId, &'static str)], edges: &[(ChannelId, ChannelId)]) -> ChannelGraph {
fn graph(
channels: &[(ChannelId, &'static str, ChannelRole)],
edges: &[(ChannelId, ChannelId)],
) -> ChannelGraph {
let mut graph = ChannelGraph {
channels: vec![],
edges: vec![],
};
for (id, name) in channels {
for (id, name, role) in channels {
graph.channels.push(Channel {
id: *id,
name: name.to_string(),
visibility: ChannelVisibility::Members,
role: *role,
})
}
@ -193,3 +197,11 @@ async fn new_test_user(db: &Arc<Database>, email: &str) -> UserId {
.unwrap()
.user_id
}
static TEST_CONNECTION_ID: AtomicU32 = AtomicU32::new(1);
fn new_test_connection(server: ServerId) -> ConnectionId {
ConnectionId {
id: TEST_CONNECTION_ID.fetch_add(1, SeqCst),
owner_id: server.0 as u32,
}
}

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

@ -1,7 +1,9 @@
use std::sync::Arc;
use crate::{
db::{
queries::channels::ChannelGraph,
tests::{graph, new_test_user, TEST_RELEASE_CHANNEL},
tests::{graph, new_test_connection, new_test_user, TEST_RELEASE_CHANNEL},
ChannelId, ChannelRole, Database, NewUserParams, RoomId,
},
test_both_dbs,
@ -11,36 +13,12 @@ use rpc::{
proto::{self},
ConnectionId,
};
use std::sync::Arc;
test_both_dbs!(test_channels, test_channels_postgres, test_channels_sqlite);
async fn test_channels(db: &Arc<Database>) {
let a_id = db
.create_user(
"user1@example.com",
false,
NewUserParams {
github_login: "user1".into(),
github_user_id: 5,
},
)
.await
.unwrap()
.user_id;
let b_id = db
.create_user(
"user2@example.com",
false,
NewUserParams {
github_login: "user2".into(),
github_user_id: 6,
},
)
.await
.unwrap()
.user_id;
let a_id = new_test_user(db, "user1@example.com").await;
let b_id = new_test_user(db, "user2@example.com").await;
let zed_id = db.create_root_channel("zed", a_id).await.unwrap();
@ -55,28 +33,28 @@ async fn test_channels(db: &Arc<Database>) {
.await
.unwrap();
let crdb_id = db.create_channel("crdb", Some(zed_id), a_id).await.unwrap();
let crdb_id = db.create_sub_channel("crdb", zed_id, a_id).await.unwrap();
let livestreaming_id = db
.create_channel("livestreaming", Some(zed_id), a_id)
.create_sub_channel("livestreaming", zed_id, a_id)
.await
.unwrap();
let replace_id = db
.create_channel("replace", Some(zed_id), a_id)
.create_sub_channel("replace", zed_id, a_id)
.await
.unwrap();
let mut members = db.get_channel_members(replace_id).await.unwrap();
let mut members = db
.transaction(|tx| async move { Ok(db.get_channel_participants(replace_id, &*tx).await?) })
.await
.unwrap();
members.sort();
assert_eq!(members, &[a_id, b_id]);
let rust_id = db.create_root_channel("rust", a_id).await.unwrap();
let cargo_id = db
.create_channel("cargo", Some(rust_id), a_id)
.await
.unwrap();
let cargo_id = db.create_sub_channel("cargo", rust_id, a_id).await.unwrap();
let cargo_ra_id = db
.create_channel("cargo-ra", Some(cargo_id), a_id)
.create_sub_channel("cargo-ra", cargo_id, a_id)
.await
.unwrap();
@ -85,13 +63,13 @@ async fn test_channels(db: &Arc<Database>) {
result.channels,
graph(
&[
(zed_id, "zed"),
(crdb_id, "crdb"),
(livestreaming_id, "livestreaming"),
(replace_id, "replace"),
(rust_id, "rust"),
(cargo_id, "cargo"),
(cargo_ra_id, "cargo-ra")
(zed_id, "zed", ChannelRole::Admin),
(crdb_id, "crdb", ChannelRole::Admin),
(livestreaming_id, "livestreaming", ChannelRole::Admin),
(replace_id, "replace", ChannelRole::Admin),
(rust_id, "rust", ChannelRole::Admin),
(cargo_id, "cargo", ChannelRole::Admin),
(cargo_ra_id, "cargo-ra", ChannelRole::Admin)
],
&[
(crdb_id, zed_id),
@ -108,10 +86,10 @@ async fn test_channels(db: &Arc<Database>) {
result.channels,
graph(
&[
(zed_id, "zed"),
(crdb_id, "crdb"),
(livestreaming_id, "livestreaming"),
(replace_id, "replace")
(zed_id, "zed", ChannelRole::Member),
(crdb_id, "crdb", ChannelRole::Member),
(livestreaming_id, "livestreaming", ChannelRole::Member),
(replace_id, "replace", ChannelRole::Member)
],
&[
(crdb_id, zed_id),
@ -136,10 +114,10 @@ async fn test_channels(db: &Arc<Database>) {
result.channels,
graph(
&[
(zed_id, "zed"),
(crdb_id, "crdb"),
(livestreaming_id, "livestreaming"),
(replace_id, "replace")
(zed_id, "zed", ChannelRole::Admin),
(crdb_id, "crdb", ChannelRole::Admin),
(livestreaming_id, "livestreaming", ChannelRole::Admin),
(replace_id, "replace", ChannelRole::Admin)
],
&[
(crdb_id, zed_id),
@ -173,35 +151,13 @@ test_both_dbs!(
async fn test_joining_channels(db: &Arc<Database>) {
let owner_id = db.create_server("test").await.unwrap().0 as u32;
let user_1 = db
.create_user(
"user1@example.com",
false,
NewUserParams {
github_login: "user1".into(),
github_user_id: 5,
},
)
.await
.unwrap()
.user_id;
let user_2 = db
.create_user(
"user2@example.com",
false,
NewUserParams {
github_login: "user2".into(),
github_user_id: 6,
},
)
.await
.unwrap()
.user_id;
let user_1 = new_test_user(db, "user1@example.com").await;
let user_2 = new_test_user(db, "user2@example.com").await;
let channel_1 = db.create_root_channel("channel_1", user_1).await.unwrap();
// can join a room with membership to its channel
let (joined_room, _) = db
let (joined_room, _, _) = db
.join_channel(
channel_1,
user_1,
@ -305,7 +261,7 @@ async fn test_channel_invites(db: &Arc<Database>) {
.unwrap();
let channel_1_3 = db
.create_channel("channel_3", Some(channel_1_1), user_1)
.create_sub_channel("channel_3", channel_1_1, user_1)
.await
.unwrap();
@ -318,7 +274,7 @@ async fn test_channel_invites(db: &Arc<Database>) {
&[
proto::ChannelMember {
user_id: user_1.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
kind: proto::channel_member::Kind::AncestorMember.into(),
role: proto::ChannelRole::Admin.into(),
},
proto::ChannelMember {
@ -407,20 +363,17 @@ async fn test_db_channel_moving(db: &Arc<Database>) {
let zed_id = db.create_root_channel("zed", a_id).await.unwrap();
let crdb_id = db.create_channel("crdb", Some(zed_id), a_id).await.unwrap();
let crdb_id = db.create_sub_channel("crdb", zed_id, a_id).await.unwrap();
let gpui2_id = db
.create_channel("gpui2", Some(zed_id), a_id)
.await
.unwrap();
let gpui2_id = db.create_sub_channel("gpui2", zed_id, a_id).await.unwrap();
let livestreaming_id = db
.create_channel("livestreaming", Some(crdb_id), a_id)
.create_sub_channel("livestreaming", crdb_id, a_id)
.await
.unwrap();
let livestreaming_dag_id = db
.create_channel("livestreaming_dag", Some(livestreaming_id), a_id)
.create_sub_channel("livestreaming_dag", livestreaming_id, a_id)
.await
.unwrap();
@ -447,299 +400,311 @@ async fn test_db_channel_moving(db: &Arc<Database>) {
.await
.is_err());
// ========================================================================
// Make a link
db.link_channel(a_id, livestreaming_id, zed_id)
.await
.unwrap();
// // ========================================================================
// // Make a link
// db.link_channel(a_id, livestreaming_id, zed_id)
// .await
// .unwrap();
// DAG is now:
// /- gpui2
// zed -- crdb - livestreaming - livestreaming_dag
// \---------/
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(gpui2_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_id, Some(crdb_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
],
);
// // DAG is now:
// // /- gpui2
// // zed -- crdb - livestreaming - livestreaming_dag
// // \---------/
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (gpui2_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_id, Some(crdb_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// ],
// );
// ========================================================================
// Create a new channel below a channel with multiple parents
let livestreaming_dag_sub_id = db
.create_channel("livestreaming_dag_sub", Some(livestreaming_dag_id), a_id)
.await
.unwrap();
// // ========================================================================
// // Create a new channel below a channel with multiple parents
// let livestreaming_dag_sub_id = db
// .create_channel("livestreaming_dag_sub", Some(livestreaming_dag_id), a_id)
// .await
// .unwrap();
// DAG is now:
// /- gpui2
// zed -- crdb - livestreaming - livestreaming_dag - livestreaming_dag_sub_id
// \---------/
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(gpui2_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_id, Some(crdb_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// // DAG is now:
// // /- gpui2
// // zed -- crdb - livestreaming - livestreaming_dag - livestreaming_dag_sub_id
// // \---------/
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (gpui2_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_id, Some(crdb_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// Test a complex DAG by making another link
let returned_channels = db
.link_channel(a_id, livestreaming_dag_sub_id, livestreaming_id)
.await
.unwrap();
// // ========================================================================
// // Test a complex DAG by making another link
// let returned_channels = db
// .link_channel(a_id, livestreaming_dag_sub_id, livestreaming_id)
// .await
// .unwrap();
// DAG is now:
// /- gpui2 /---------------------\
// zed - crdb - livestreaming - livestreaming_dag - livestreaming_dag_sub_id
// \--------/
// // DAG is now:
// // /- gpui2 /---------------------\
// // zed - crdb - livestreaming - livestreaming_dag - livestreaming_dag_sub_id
// // \--------/
// make sure we're getting just the new link
// Not using the assert_dag helper because we want to make sure we're returning the full data
pretty_assertions::assert_eq!(
returned_channels,
graph(
&[(livestreaming_dag_sub_id, "livestreaming_dag_sub")],
&[(livestreaming_dag_sub_id, livestreaming_id)]
)
);
// // make sure we're getting just the new link
// // Not using the assert_dag helper because we want to make sure we're returning the full data
// pretty_assertions::assert_eq!(
// returned_channels,
// graph(
// &[(
// livestreaming_dag_sub_id,
// "livestreaming_dag_sub",
// ChannelRole::Admin
// )],
// &[(livestreaming_dag_sub_id, livestreaming_id)]
// )
// );
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(gpui2_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_id, Some(crdb_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (gpui2_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_id, Some(crdb_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// Test a complex DAG by making another link
let returned_channels = db
.link_channel(a_id, livestreaming_id, gpui2_id)
.await
.unwrap();
// // ========================================================================
// // Test a complex DAG by making another link
// let returned_channels = db
// .link_channel(a_id, livestreaming_id, gpui2_id)
// .await
// .unwrap();
// DAG is now:
// /- gpui2 -\ /---------------------\
// zed - crdb -- livestreaming - livestreaming_dag - livestreaming_dag_sub_id
// \---------/
// // DAG is now:
// // /- gpui2 -\ /---------------------\
// // zed - crdb -- livestreaming - livestreaming_dag - livestreaming_dag_sub_id
// // \---------/
// Make sure that we're correctly getting the full sub-dag
pretty_assertions::assert_eq!(
returned_channels,
graph(
&[
(livestreaming_id, "livestreaming"),
(livestreaming_dag_id, "livestreaming_dag"),
(livestreaming_dag_sub_id, "livestreaming_dag_sub"),
],
&[
(livestreaming_id, gpui2_id),
(livestreaming_dag_id, livestreaming_id),
(livestreaming_dag_sub_id, livestreaming_id),
(livestreaming_dag_sub_id, livestreaming_dag_id),
]
)
);
// // Make sure that we're correctly getting the full sub-dag
// pretty_assertions::assert_eq!(
// returned_channels,
// graph(
// &[
// (livestreaming_id, "livestreaming", ChannelRole::Admin),
// (
// livestreaming_dag_id,
// "livestreaming_dag",
// ChannelRole::Admin
// ),
// (
// livestreaming_dag_sub_id,
// "livestreaming_dag_sub",
// ChannelRole::Admin
// ),
// ],
// &[
// (livestreaming_id, gpui2_id),
// (livestreaming_dag_id, livestreaming_id),
// (livestreaming_dag_sub_id, livestreaming_id),
// (livestreaming_dag_sub_id, livestreaming_dag_id),
// ]
// )
// );
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(gpui2_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_id, Some(crdb_id)),
(livestreaming_id, Some(gpui2_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (gpui2_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_id, Some(crdb_id)),
// (livestreaming_id, Some(gpui2_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// Test unlinking in a complex DAG by removing the inner link
db.unlink_channel(a_id, livestreaming_dag_sub_id, livestreaming_id)
.await
.unwrap();
// // ========================================================================
// // Test unlinking in a complex DAG by removing the inner link
// db.unlink_channel(a_id, livestreaming_dag_sub_id, livestreaming_id)
// .await
// .unwrap();
// DAG is now:
// /- gpui2 -\
// zed - crdb -- livestreaming - livestreaming_dag - livestreaming_dag_sub
// \---------/
// // DAG is now:
// // /- gpui2 -\
// // zed - crdb -- livestreaming - livestreaming_dag - livestreaming_dag_sub
// // \---------/
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(gpui2_id, Some(zed_id)),
(livestreaming_id, Some(gpui2_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_id, Some(crdb_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (gpui2_id, Some(zed_id)),
// (livestreaming_id, Some(gpui2_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_id, Some(crdb_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// Test unlinking in a complex DAG by removing the inner link
db.unlink_channel(a_id, livestreaming_id, gpui2_id)
.await
.unwrap();
// // ========================================================================
// // Test unlinking in a complex DAG by removing the inner link
// db.unlink_channel(a_id, livestreaming_id, gpui2_id)
// .await
// .unwrap();
// DAG is now:
// /- gpui2
// zed - crdb -- livestreaming - livestreaming_dag - livestreaming_dag_sub
// \---------/
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(gpui2_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_id, Some(crdb_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// // DAG is now:
// // /- gpui2
// // zed - crdb -- livestreaming - livestreaming_dag - livestreaming_dag_sub
// // \---------/
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (gpui2_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_id, Some(crdb_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// Test moving DAG nodes by moving livestreaming to be below gpui2
db.move_channel(a_id, livestreaming_id, crdb_id, gpui2_id)
.await
.unwrap();
// // ========================================================================
// // Test moving DAG nodes by moving livestreaming to be below gpui2
// db.move_channel(livestreaming_id, Some(crdb_id), gpui2_id, a_id)
// .await
// .unwrap();
// DAG is now:
// /- gpui2 -- livestreaming - livestreaming_dag - livestreaming_dag_sub
// zed - crdb /
// \---------/
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(gpui2_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_id, Some(gpui2_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// // DAG is now:
// // /- gpui2 -- livestreaming - livestreaming_dag - livestreaming_dag_sub
// // zed - crdb /
// // \---------/
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (gpui2_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_id, Some(gpui2_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// Deleting a channel should not delete children that still have other parents
db.delete_channel(gpui2_id, a_id).await.unwrap();
// // ========================================================================
// // Deleting a channel should not delete children that still have other parents
// db.delete_channel(gpui2_id, a_id).await.unwrap();
// DAG is now:
// zed - crdb
// \- livestreaming - livestreaming_dag - livestreaming_dag_sub
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// // DAG is now:
// // zed - crdb
// // \- livestreaming - livestreaming_dag - livestreaming_dag_sub
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// Unlinking a channel from it's parent should automatically promote it to a root channel
db.unlink_channel(a_id, crdb_id, zed_id).await.unwrap();
// // ========================================================================
// // Unlinking a channel from it's parent should automatically promote it to a root channel
// db.unlink_channel(a_id, crdb_id, zed_id).await.unwrap();
// DAG is now:
// crdb
// zed
// \- livestreaming - livestreaming_dag - livestreaming_dag_sub
// // DAG is now:
// // crdb
// // zed
// // \- livestreaming - livestreaming_dag - livestreaming_dag_sub
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, None),
(livestreaming_id, Some(zed_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, None),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// You should be able to move a root channel into a non-root channel
db.link_channel(a_id, crdb_id, zed_id).await.unwrap();
// // ========================================================================
// // You should be able to move a root channel into a non-root channel
// db.link_channel(a_id, crdb_id, zed_id).await.unwrap();
// DAG is now:
// zed - crdb
// \- livestreaming - livestreaming_dag - livestreaming_dag_sub
// // DAG is now:
// // zed - crdb
// // \- livestreaming - livestreaming_dag - livestreaming_dag_sub
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// ========================================================================
// Prep for DAG deletion test
db.link_channel(a_id, livestreaming_id, crdb_id)
.await
.unwrap();
// // ========================================================================
// // Prep for DAG deletion test
// db.link_channel(a_id, livestreaming_id, crdb_id)
// .await
// .unwrap();
// DAG is now:
// zed - crdb - livestreaming - livestreaming_dag - livestreaming_dag_sub
// \--------/
// // DAG is now:
// // zed - crdb - livestreaming - livestreaming_dag - livestreaming_dag_sub
// // \--------/
let result = db.get_channels_for_user(a_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(crdb_id, Some(zed_id)),
(livestreaming_id, Some(zed_id)),
(livestreaming_id, Some(crdb_id)),
(livestreaming_dag_id, Some(livestreaming_id)),
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
],
);
// let result = db.get_channels_for_user(a_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (crdb_id, Some(zed_id)),
// (livestreaming_id, Some(zed_id)),
// (livestreaming_id, Some(crdb_id)),
// (livestreaming_dag_id, Some(livestreaming_id)),
// (livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
// ],
// );
// Deleting the parent of a DAG should delete the whole DAG:
db.delete_channel(zed_id, a_id).await.unwrap();
let result = db.get_channels_for_user(a_id).await.unwrap();
// // Deleting the parent of a DAG should delete the whole DAG:
// db.delete_channel(zed_id, a_id).await.unwrap();
// let result = db.get_channels_for_user(a_id).await.unwrap();
assert!(result.channels.is_empty())
// assert!(result.channels.is_empty())
}
test_both_dbs!(
@ -765,12 +730,12 @@ async fn test_db_channel_moving_bugs(db: &Arc<Database>) {
let zed_id = db.create_root_channel("zed", user_id).await.unwrap();
let projects_id = db
.create_channel("projects", Some(zed_id), user_id)
.create_sub_channel("projects", zed_id, user_id)
.await
.unwrap();
let livestreaming_id = db
.create_channel("livestreaming", Some(projects_id), user_id)
.create_sub_channel("livestreaming", projects_id, user_id)
.await
.unwrap();
@ -778,25 +743,37 @@ async fn test_db_channel_moving_bugs(db: &Arc<Database>) {
// Move to same parent should be a no-op
assert!(db
.move_channel(user_id, projects_id, zed_id, zed_id)
.move_channel(projects_id, Some(zed_id), zed_id, user_id)
.await
.unwrap()
.is_empty());
// Stranding a channel should retain it's sub channels
db.unlink_channel(user_id, projects_id, zed_id)
.await
.unwrap();
.is_none());
let result = db.get_channels_for_user(user_id).await.unwrap();
assert_dag(
result.channels,
&[
(zed_id, None),
(projects_id, None),
(projects_id, Some(zed_id)),
(livestreaming_id, Some(projects_id)),
],
);
// Stranding a channel should retain it's sub channels
// Commented out as we don't fix permissions when this happens yet.
//
// db.unlink_channel(user_id, projects_id, zed_id)
// .await
// .unwrap();
// let result = db.get_channels_for_user(user_id).await.unwrap();
// assert_dag(
// result.channels,
// &[
// (zed_id, None),
// (projects_id, None),
// (livestreaming_id, Some(projects_id)),
// ],
// );
}
test_both_dbs!(
@ -812,11 +789,11 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
let zed_channel = db.create_root_channel("zed", admin).await.unwrap();
let active_channel = db
.create_channel("active", Some(zed_channel), admin)
.create_sub_channel("active", zed_channel, admin)
.await
.unwrap();
let vim_channel = db
.create_channel("vim", Some(active_channel), admin)
.create_sub_channel("vim", active_channel, admin)
.await
.unwrap();
@ -859,7 +836,7 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
&[
proto::ChannelMember {
user_id: admin.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
kind: proto::channel_member::Kind::AncestorMember.into(),
role: proto::ChannelRole::Admin.into(),
},
proto::ChannelMember {
@ -917,7 +894,7 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
&[
proto::ChannelMember {
user_id: admin.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
kind: proto::channel_member::Kind::AncestorMember.into(),
role: proto::ChannelRole::Admin.into(),
},
proto::ChannelMember {
@ -958,7 +935,7 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
&[
proto::ChannelMember {
user_id: admin.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
kind: proto::channel_member::Kind::AncestorMember.into(),
role: proto::ChannelRole::Admin.into(),
},
proto::ChannelMember {
@ -1006,7 +983,7 @@ async fn test_user_is_channel_participant(db: &Arc<Database>) {
&[
proto::ChannelMember {
user_id: admin.to_proto(),
kind: proto::channel_member::Kind::Member.into(),
kind: proto::channel_member::Kind::AncestorMember.into(),
role: proto::ChannelRole::Admin.into(),
},
proto::ChannelMember {
@ -1041,17 +1018,17 @@ async fn test_user_joins_correct_channel(db: &Arc<Database>) {
let zed_channel = db.create_root_channel("zed", admin).await.unwrap();
let active_channel = db
.create_channel("active", Some(zed_channel), admin)
.create_sub_channel("active", zed_channel, admin)
.await
.unwrap();
let vim_channel = db
.create_channel("vim", Some(active_channel), admin)
.create_sub_channel("vim", active_channel, admin)
.await
.unwrap();
let vim2_channel = db
.create_channel("vim2", Some(vim_channel), admin)
.create_sub_channel("vim2", vim_channel, admin)
.await
.unwrap();
@ -1068,15 +1045,52 @@ async fn test_user_joins_correct_channel(db: &Arc<Database>) {
.unwrap();
let most_public = db
.transaction(
|tx| async move { db.most_public_ancestor_for_channel(vim_channel, &*tx).await },
)
.transaction(|tx| async move {
Ok(db
.public_path_to_channel(vim_channel, &tx)
.await?
.first()
.cloned())
})
.await
.unwrap();
assert_eq!(most_public, Some(zed_channel))
}
test_both_dbs!(
test_guest_access,
test_guest_access_postgres,
test_guest_access_sqlite
);
async fn test_guest_access(db: &Arc<Database>) {
let server = db.create_server("test").await.unwrap();
let admin = new_test_user(db, "admin@example.com").await;
let guest = new_test_user(db, "guest@example.com").await;
let guest_connection = new_test_connection(server);
let zed_channel = db.create_root_channel("zed", admin).await.unwrap();
db.set_channel_visibility(zed_channel, crate::db::ChannelVisibility::Public, admin)
.await
.unwrap();
assert!(db
.join_channel_chat(zed_channel, guest_connection, guest)
.await
.is_err());
db.join_channel(zed_channel, guest, guest_connection, TEST_RELEASE_CHANNEL)
.await
.unwrap();
assert!(db
.join_channel_chat(zed_channel, guest_connection, guest)
.await
.is_ok())
}
#[track_caller]
fn assert_dag(actual: ChannelGraph, expected: &[(ChannelId, Option<ChannelId>)]) {
let mut actual_map: HashMap<ChannelId, HashSet<ChannelId>> = HashMap::default();

31
crates/collab/src/db/tests/message_tests.rs
View file

@ -15,18 +15,22 @@ test_both_dbs!(
async fn test_channel_message_retrieval(db: &Arc<Database>) {
let user = new_test_user(db, "user@example.com").await;
let channel = db.create_channel("channel", None, user).await.unwrap();
let result = db.create_channel("channel", None, user).await.unwrap();
let owner_id = db.create_server("test").await.unwrap().0 as u32;
db.join_channel_chat(channel, rpc::ConnectionId { owner_id, id: 0 }, user)
.await
.unwrap();
db.join_channel_chat(
result.channel.id,
rpc::ConnectionId { owner_id, id: 0 },
user,
)
.await
.unwrap();
let mut all_messages = Vec::new();
for i in 0..10 {
all_messages.push(
db.create_channel_message(
channel,
result.channel.id,
user,
&i.to_string(),
&[],
@ -41,7 +45,7 @@ async fn test_channel_message_retrieval(db: &Arc<Database>) {
}
let messages = db
.get_channel_messages(channel, user, 3, None)
.get_channel_messages(result.channel.id, user, 3, None)
.await
.unwrap()
.into_iter()
@ -51,7 +55,7 @@ async fn test_channel_message_retrieval(db: &Arc<Database>) {
let messages = db
.get_channel_messages(
channel,
result.channel.id,
user,
4,
Some(MessageId::from_proto(all_messages[6])),
@ -74,7 +78,7 @@ async fn test_channel_message_nonces(db: &Arc<Database>) {
let user_a = new_test_user(db, "user_a@example.com").await;
let user_b = new_test_user(db, "user_b@example.com").await;
let user_c = new_test_user(db, "user_c@example.com").await;
let channel = db.create_channel("channel", None, user_a).await.unwrap();
let channel = db.create_root_channel("channel", user_a).await.unwrap();
db.invite_channel_member(channel, user_b, user_a, ChannelRole::Member)
.await
.unwrap();
@ -206,8 +210,8 @@ async fn test_unseen_channel_messages(db: &Arc<Database>) {
let user = new_test_user(db, "user_a@example.com").await;
let observer = new_test_user(db, "user_b@example.com").await;
let channel_1 = db.create_channel("channel", None, user).await.unwrap();
let channel_2 = db.create_channel("channel-2", None, user).await.unwrap();
let channel_1 = db.create_root_channel("channel", user).await.unwrap();
let channel_2 = db.create_root_channel("channel-2", user).await.unwrap();
db.invite_channel_member(channel_1, observer, user, ChannelRole::Member)
.await
@ -362,7 +366,12 @@ async fn test_channel_message_mentions(db: &Arc<Database>) {
let user_b = new_test_user(db, "user_b@example.com").await;
let user_c = new_test_user(db, "user_c@example.com").await;
let channel = db.create_channel("channel", None, user_a).await.unwrap();
let channel = db
.create_channel("channel", None, user_a)
.await
.unwrap()
.channel
.id;
db.invite_channel_member(channel, user_b, user_a, ChannelRole::Member)
.await
.unwrap();

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

@ -3,8 +3,11 @@ mod connection_pool;
use crate::{
auth,
db::{
self, BufferId, ChannelId, ChannelVisibility, ChannelsForUser, CreatedChannelMessage,
Database, MessageId, NotificationId, ProjectId, RoomId, ServerId, User, UserId,
self, BufferId, ChannelId, ChannelRole, ChannelsForUser, CreateChannelResult,
CreatedChannelMessage, Database, InviteMemberResult, MembershipUpdated, MessageId,
MoveChannelResult, NotificationId, ProjectId, RemoveChannelMemberResult,
RenameChannelResult, RespondToChannelInvite, RoomId, ServerId, SetChannelVisibilityResult,
User, UserId,
},
executor::Executor,
AppState, Result,
@ -38,8 +41,8 @@ use lazy_static::lazy_static;
use prometheus::{register_int_gauge, IntGauge};
use rpc::{
proto::{
self, Ack, AnyTypedEnvelope, ChannelEdge, EntityMessage, EnvelopedMessage,
LiveKitConnectionInfo, RequestMessage, UpdateChannelBufferCollaborators,
self, Ack, AnyTypedEnvelope, EntityMessage, EnvelopedMessage, LiveKitConnectionInfo,
RequestMessage, UpdateChannelBufferCollaborators,
},
Connection, ConnectionId, Peer, Receipt, TypedEnvelope,
};
@ -594,7 +597,7 @@ impl Server {
let mut pool = this.connection_pool.lock();
pool.add_connection(connection_id, user_id, user.admin);
this.peer.send(connection_id, build_initial_contacts_update(contacts, &pool))?;
this.peer.send(connection_id, build_initial_channels_update(
this.peer.send(connection_id, build_channels_update(
channels_for_user,
channel_invites
))?;
@ -951,6 +954,7 @@ async fn create_room(
Some(proto::LiveKitConnectionInfo {
server_url: live_kit.url().into(),
token,
can_publish: true,
})
})
}
@ -1031,6 +1035,7 @@ async fn join_room(
Some(proto::LiveKitConnectionInfo {
server_url: live_kit.url().into(),
token,
can_publish: true,
})
} else {
None
@ -2217,38 +2222,21 @@ async fn create_channel(
let db = session.db().await;
let parent_id = request.parent_id.map(|id| ChannelId::from_proto(id));
let id = db
let CreateChannelResult {
channel,
participants_to_update,
} = db
.create_channel(&request.name, parent_id, session.user_id)
.await?;
let channel = proto::Channel {
id: id.to_proto(),
name: request.name,
visibility: proto::ChannelVisibility::Members as i32,
};
response.send(proto::CreateChannelResponse {
channel: Some(channel.clone()),
channel: Some(channel.to_proto()),
parent_id: request.parent_id,
})?;
let Some(parent_id) = parent_id else {
return Ok(());
};
let update = proto::UpdateChannels {
channels: vec![channel],
insert_edge: vec![ChannelEdge {
parent_id: parent_id.to_proto(),
channel_id: id.to_proto(),
}],
..Default::default()
};
let user_ids_to_notify = db.get_channel_members(parent_id).await?;
let connection_pool = session.connection_pool().await;
for user_id in user_ids_to_notify {
for (user_id, channels) in participants_to_update {
let update = build_channels_update(channels, vec![]);
for connection_id in connection_pool.user_connection_ids(user_id) {
if user_id == session.user_id {
continue;
@ -2297,7 +2285,10 @@ async fn invite_channel_member(
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let invitee_id = UserId::from_proto(request.user_id);
let notifications = db
let InviteMemberResult {
channel,
notifications,
} = db
.invite_channel_member(
channel_id,
invitee_id,
@ -2306,21 +2297,17 @@ async fn invite_channel_member(
)
.await?;
let channel = db.get_channel(channel_id, session.user_id).await?;
let update = proto::UpdateChannels {
channel_invitations: vec![channel.to_proto()],
..Default::default()
};
let mut update = proto::UpdateChannels::default();
update.channel_invitations.push(proto::Channel {
id: channel.id.to_proto(),
visibility: channel.visibility.into(),
name: channel.name,
});
let pool = session.connection_pool().await;
for connection_id in pool.user_connection_ids(invitee_id) {
let connection_pool = session.connection_pool().await;
for connection_id in connection_pool.user_connection_ids(invitee_id) {
session.peer.send(connection_id, update.clone())?;
}
send_notifications(&*pool, &session.peer, notifications);
send_notifications(&*connection_pool, &session.peer, notifications);
response.send(proto::Ack {})?;
Ok(())
@ -2335,20 +2322,22 @@ async fn remove_channel_member(
let channel_id = ChannelId::from_proto(request.channel_id);
let member_id = UserId::from_proto(request.user_id);
let removed_notification_id = db
let RemoveChannelMemberResult {
membership_update,
notification_id,
} = db
.remove_channel_member(channel_id, member_id, session.user_id)
.await?;
let mut update = proto::UpdateChannels::default();
update.delete_channels.push(channel_id.to_proto());
for connection_id in session
.connection_pool()
.await
.user_connection_ids(member_id)
{
session.peer.send(connection_id, update.clone()).trace_err();
if let Some(notification_id) = removed_notification_id {
let connection_pool = &session.connection_pool().await;
notify_membership_updated(
&connection_pool,
membership_update,
member_id,
&session.peer,
);
for connection_id in connection_pool.user_connection_ids(member_id) {
if let Some(notification_id) = notification_id {
session
.peer
.send(
@ -2374,22 +2363,27 @@ async fn set_channel_visibility(
let channel_id = ChannelId::from_proto(request.channel_id);
let visibility = request.visibility().into();
let channel = db
let SetChannelVisibilityResult {
participants_to_update,
participants_to_remove,
channels_to_remove,
} = db
.set_channel_visibility(channel_id, visibility, session.user_id)
.await?;
let mut update = proto::UpdateChannels::default();
update.channels.push(proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
visibility: channel.visibility.into(),
});
let member_ids = db.get_channel_members(channel_id).await?;
let connection_pool = session.connection_pool().await;
for member_id in member_ids {
for connection_id in connection_pool.user_connection_ids(member_id) {
for (user_id, channels) in participants_to_update {
let update = build_channels_update(channels, vec![]);
for connection_id in connection_pool.user_connection_ids(user_id) {
session.peer.send(connection_id, update.clone())?;
}
}
for user_id in participants_to_remove {
let update = proto::UpdateChannels {
delete_channels: channels_to_remove.iter().map(|id| id.to_proto()).collect(),
..Default::default()
};
for connection_id in connection_pool.user_connection_ids(user_id) {
session.peer.send(connection_id, update.clone())?;
}
}
@ -2406,7 +2400,7 @@ async fn set_channel_member_role(
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let member_id = UserId::from_proto(request.user_id);
let channel_member = db
let result = db
.set_channel_member_role(
channel_id,
session.user_id,
@ -2415,22 +2409,30 @@ async fn set_channel_member_role(
)
.await?;
let channel = db.get_channel(channel_id, session.user_id).await?;
match result {
db::SetMemberRoleResult::MembershipUpdated(membership_update) => {
let connection_pool = session.connection_pool().await;
notify_membership_updated(
&connection_pool,
membership_update,
member_id,
&session.peer,
)
}
db::SetMemberRoleResult::InviteUpdated(channel) => {
let update = proto::UpdateChannels {
channel_invitations: vec![channel.to_proto()],
..Default::default()
};
let mut update = proto::UpdateChannels::default();
if channel_member.accepted {
update.channel_permissions.push(proto::ChannelPermission {
channel_id: channel.id.to_proto(),
role: request.role,
});
}
for connection_id in session
.connection_pool()
.await
.user_connection_ids(member_id)
{
session.peer.send(connection_id, update.clone())?;
for connection_id in session
.connection_pool()
.await
.user_connection_ids(member_id)
{
session.peer.send(connection_id, update.clone())?;
}
}
}
response.send(proto::Ack {})?;
@ -2444,26 +2446,25 @@ async fn rename_channel(
) -> Result<()> {
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let channel = db
let RenameChannelResult {
channel,
participants_to_update,
} = db
.rename_channel(channel_id, session.user_id, &request.name)
.await?;
let channel = proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
visibility: channel.visibility.into(),
};
response.send(proto::RenameChannelResponse {
channel: Some(channel.clone()),
channel: Some(channel.to_proto()),
})?;
let mut update = proto::UpdateChannels::default();
update.channels.push(channel);
let member_ids = db.get_channel_members(channel_id).await?;
let connection_pool = session.connection_pool().await;
for member_id in member_ids {
for connection_id in connection_pool.user_connection_ids(member_id) {
for (user_id, channel) in participants_to_update {
for connection_id in connection_pool.user_connection_ids(user_id) {
let update = proto::UpdateChannels {
channels: vec![channel.to_proto()],
..Default::default()
};
session.peer.send(connection_id, update.clone())?;
}
}
@ -2471,6 +2472,8 @@ async fn rename_channel(
Ok(())
}
// TODO: Implement in terms of symlinks
// Current behavior of this is more like 'Move root channel'
async fn link_channel(
request: proto::LinkChannel,
response: Response<proto::LinkChannel>,
@ -2479,64 +2482,26 @@ async fn link_channel(
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let to = ChannelId::from_proto(request.to);
let channels_to_send = db.link_channel(session.user_id, channel_id, to).await?;
let members = db.get_channel_members(to).await?;
let connection_pool = session.connection_pool().await;
let update = proto::UpdateChannels {
channels: channels_to_send
.channels
.into_iter()
.map(|channel| proto::Channel {
id: channel.id.to_proto(),
visibility: channel.visibility.into(),
name: channel.name,
})
.collect(),
insert_edge: channels_to_send.edges,
..Default::default()
};
for member_id in members {
for connection_id in connection_pool.user_connection_ids(member_id) {
session.peer.send(connection_id, update.clone())?;
}
}
let result = db
.move_channel(channel_id, None, to, session.user_id)
.await?;
drop(db);
notify_channel_moved(result, session).await?;
response.send(Ack {})?;
Ok(())
}
// TODO: Implement in terms of symlinks
async fn unlink_channel(
request: proto::UnlinkChannel,
response: Response<proto::UnlinkChannel>,
session: Session,
_request: proto::UnlinkChannel,
_response: Response<proto::UnlinkChannel>,
_session: Session,
) -> Result<()> {
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let from = ChannelId::from_proto(request.from);
db.unlink_channel(session.user_id, channel_id, from).await?;
let members = db.get_channel_members(from).await?;
let update = proto::UpdateChannels {
delete_edge: vec![proto::ChannelEdge {
channel_id: channel_id.to_proto(),
parent_id: from.to_proto(),
}],
..Default::default()
};
let connection_pool = session.connection_pool().await;
for member_id in members {
for connection_id in connection_pool.user_connection_ids(member_id) {
session.peer.send(connection_id, update.clone())?;
}
}
response.send(Ack {})?;
Ok(())
Err(anyhow!("unimplemented").into())
}
async fn move_channel(
@ -2549,53 +2514,46 @@ async fn move_channel(
let from_parent = ChannelId::from_proto(request.from);
let to = ChannelId::from_proto(request.to);
let channels_to_send = db
.move_channel(session.user_id, channel_id, from_parent, to)
let result = db
.move_channel(channel_id, Some(from_parent), to, session.user_id)
.await?;
drop(db);
if channels_to_send.is_empty() {
response.send(Ack {})?;
return Ok(());
}
let members_from = db.get_channel_members(from_parent).await?;
let members_to = db.get_channel_members(to).await?;
let update = proto::UpdateChannels {
delete_edge: vec![proto::ChannelEdge {
channel_id: channel_id.to_proto(),
parent_id: from_parent.to_proto(),
}],
..Default::default()
};
let connection_pool = session.connection_pool().await;
for member_id in members_from {
for connection_id in connection_pool.user_connection_ids(member_id) {
session.peer.send(connection_id, update.clone())?;
}
}
let update = proto::UpdateChannels {
channels: channels_to_send
.channels
.into_iter()
.map(|channel| proto::Channel {
id: channel.id.to_proto(),
visibility: channel.visibility.into(),
name: channel.name,
})
.collect(),
insert_edge: channels_to_send.edges,
..Default::default()
};
for member_id in members_to {
for connection_id in connection_pool.user_connection_ids(member_id) {
session.peer.send(connection_id, update.clone())?;
}
}
notify_channel_moved(result, session).await?;
response.send(Ack {})?;
Ok(())
}
async fn notify_channel_moved(result: Option<MoveChannelResult>, session: Session) -> Result<()> {
let Some(MoveChannelResult {
participants_to_remove,
participants_to_update,
moved_channels,
}) = result
else {
return Ok(());
};
let moved_channels: Vec<u64> = moved_channels.iter().map(|id| id.to_proto()).collect();
let connection_pool = session.connection_pool().await;
for (user_id, channels) in participants_to_update {
let mut update = build_channels_update(channels, vec![]);
update.delete_channels = moved_channels.clone();
for connection_id in connection_pool.user_connection_ids(user_id) {
session.peer.send(connection_id, update.clone())?;
}
}
for user_id in participants_to_remove {
let update = proto::UpdateChannels {
delete_channels: moved_channels.clone(),
..Default::default()
};
for connection_id in connection_pool.user_connection_ids(user_id) {
session.peer.send(connection_id, update.clone())?;
}
}
Ok(())
}
@ -2620,81 +2578,39 @@ async fn respond_to_channel_invite(
) -> Result<()> {
let db = session.db().await;
let channel_id = ChannelId::from_proto(request.channel_id);
let notifications = db
let RespondToChannelInvite {
membership_update,
notifications,
} = db
.respond_to_channel_invite(channel_id, session.user_id, request.accept)
.await?;
if request.accept {
channel_membership_updated(db, channel_id, &session).await?;
let connection_pool = session.connection_pool().await;
if let Some(membership_update) = membership_update {
notify_membership_updated(
&connection_pool,
membership_update,
session.user_id,
&session.peer,
);
} else {
let mut update = proto::UpdateChannels::default();
update
.remove_channel_invitations
.push(channel_id.to_proto());
session.peer.send(session.connection_id, update)?;
}
let update = proto::UpdateChannels {
remove_channel_invitations: vec![channel_id.to_proto()],
..Default::default()
};
for connection_id in connection_pool.user_connection_ids(session.user_id) {
session.peer.send(connection_id, update.clone())?;
}
};
send_notifications(&*connection_pool, &session.peer, notifications);
send_notifications(
&*session.connection_pool().await,
&session.peer,
notifications,
);
response.send(proto::Ack {})?;
Ok(())
}
async fn channel_membership_updated(
db: tokio::sync::MutexGuard<'_, DbHandle>,
channel_id: ChannelId,
session: &Session,
) -> Result<(), crate::Error> {
let mut update = proto::UpdateChannels::default();
update
.remove_channel_invitations
.push(channel_id.to_proto());
let result = db.get_channel_for_user(channel_id, session.user_id).await?;
update.channels.extend(
result
.channels
.channels
.into_iter()
.map(|channel| proto::Channel {
id: channel.id.to_proto(),
visibility: channel.visibility.into(),
name: channel.name,
}),
);
update.unseen_channel_messages = result.channel_messages;
update.unseen_channel_buffer_changes = result.unseen_buffer_changes;
update.insert_edge = result.channels.edges;
update
.channel_participants
.extend(
result
.channel_participants
.into_iter()
.map(|(channel_id, user_ids)| proto::ChannelParticipants {
channel_id: channel_id.to_proto(),
participant_user_ids: user_ids.into_iter().map(UserId::to_proto).collect(),
}),
);
update
.channel_permissions
.extend(
result
.channels_with_admin_privileges
.into_iter()
.map(|channel_id| proto::ChannelPermission {
channel_id: channel_id.to_proto(),
role: proto::ChannelRole::Admin.into(),
}),
);
session.peer.send(session.connection_id, update)?;
Ok(())
}
async fn join_channel(
request: proto::JoinChannel,
response: Response<proto::JoinChannel>,
@ -2727,7 +2643,7 @@ async fn join_channel_internal(
leave_room_for_session(&session).await?;
let db = session.db().await;
let (joined_room, joined_channel) = db
let (joined_room, membership_updated, role) = db
.join_channel(
channel_id,
session.user_id,
@ -2737,16 +2653,32 @@ async fn join_channel_internal(
.await?;
let live_kit_connection_info = session.live_kit_client.as_ref().and_then(|live_kit| {
let token = live_kit
.room_token(
&joined_room.room.live_kit_room,
&session.user_id.to_string(),
let (can_publish, token) = if role == ChannelRole::Guest {
(
false,
live_kit
.guest_token(
&joined_room.room.live_kit_room,
&session.user_id.to_string(),
)
.trace_err()?,
)
.trace_err()?;
} else {
(
true,
live_kit
.room_token(
&joined_room.room.live_kit_room,
&session.user_id.to_string(),
)
.trace_err()?,
)
};
Some(LiveKitConnectionInfo {
server_url: live_kit.url().into(),
token,
can_publish,
})
});
@ -2756,8 +2688,14 @@ async fn join_channel_internal(
live_kit_connection_info,
})?;
if let Some(joined_channel) = joined_channel {
channel_membership_updated(db, joined_channel, &session).await?
let connection_pool = session.connection_pool().await;
if let Some(membership_updated) = membership_updated {
notify_membership_updated(
&connection_pool,
membership_updated,
session.user_id,
&session.peer,
);
}
room_updated(&joined_room.room, &session.peer);
@ -3281,7 +3219,26 @@ fn to_tungstenite_message(message: AxumMessage) -> TungsteniteMessage {
}
}
fn build_initial_channels_update(
fn notify_membership_updated(
connection_pool: &ConnectionPool,
result: MembershipUpdated,
user_id: UserId,
peer: &Peer,
) {
let mut update = build_channels_update(result.new_channels, vec![]);
update.delete_channels = result
.removed_channels
.into_iter()
.map(|id| id.to_proto())
.collect();
update.remove_channel_invitations = vec![result.channel_id.to_proto()];
for connection_id in connection_pool.user_connection_ids(user_id) {
peer.send(connection_id, update.clone()).trace_err();
}
}
fn build_channels_update(
channels: ChannelsForUser,
channel_invites: Vec<db::Channel>,
) -> proto::UpdateChannels {
@ -3292,6 +3249,7 @@ fn build_initial_channels_update(
id: channel.id.to_proto(),
name: channel.name,
visibility: channel.visibility.into(),
role: channel.role.into(),
});
}
@ -3308,24 +3266,12 @@ fn build_initial_channels_update(
});
}
update
.channel_permissions
.extend(
channels
.channels_with_admin_privileges
.into_iter()
.map(|id| proto::ChannelPermission {
channel_id: id.to_proto(),
role: proto::ChannelRole::Admin.into(),
}),
);
for channel in channel_invites {
update.channel_invitations.push(proto::Channel {
id: channel.id.to_proto(),
name: channel.name,
// TODO: Visibility
visibility: ChannelVisibility::Public.into(),
visibility: channel.visibility.into(),
role: channel.role.into(),
});
}

24
crates/collab/src/tests/channel_buffer_tests.rs
View file

@ -410,10 +410,10 @@ async fn test_channel_buffer_disconnect(
server.disconnect_client(client_a.peer_id().unwrap());
deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT);
channel_buffer_a.update(cx_a, |buffer, _| {
channel_buffer_a.update(cx_a, |buffer, cx| {
assert_eq!(
buffer.channel().as_ref(),
&channel(channel_id, "the-channel")
buffer.channel(cx).unwrap().as_ref(),
&channel(channel_id, "the-channel", proto::ChannelRole::Admin)
);
assert!(!buffer.is_connected());
});
@ -435,18 +435,16 @@ async fn test_channel_buffer_disconnect(
deterministic.run_until_parked();
// Channel buffer observed the deletion
channel_buffer_b.update(cx_b, |buffer, _| {
assert_eq!(
buffer.channel().as_ref(),
&channel(channel_id, "the-channel")
);
channel_buffer_b.update(cx_b, |buffer, cx| {
assert!(buffer.channel(cx).is_none());
assert!(!buffer.is_connected());
});
}
fn channel(id: u64, name: &'static str) -> Channel {
fn channel(id: u64, name: &'static str, role: proto::ChannelRole) -> Channel {
Channel {
id,
role,
name: name.to_string(),
visibility: proto::ChannelVisibility::Members,
unseen_note_version: None,
@ -698,7 +696,7 @@ async fn test_following_to_channel_notes_without_a_shared_project(
.await
.unwrap();
channel_view_1_a.update(cx_a, |notes, cx| {
assert_eq!(notes.channel(cx).name, "channel-1");
assert_eq!(notes.channel(cx).unwrap().name, "channel-1");
notes.editor.update(cx, |editor, cx| {
editor.insert("Hello from A.", cx);
editor.change_selections(None, cx, |selections| {
@ -730,7 +728,7 @@ async fn test_following_to_channel_notes_without_a_shared_project(
.expect("active item is not a channel view")
});
channel_view_1_b.read_with(cx_b, |notes, cx| {
assert_eq!(notes.channel(cx).name, "channel-1");
assert_eq!(notes.channel(cx).unwrap().name, "channel-1");
let editor = notes.editor.read(cx);
assert_eq!(editor.text(cx), "Hello from A.");
assert_eq!(editor.selections.ranges::<usize>(cx), &[3..4]);
@ -742,7 +740,7 @@ async fn test_following_to_channel_notes_without_a_shared_project(
.await
.unwrap();
channel_view_2_a.read_with(cx_a, |notes, cx| {
assert_eq!(notes.channel(cx).name, "channel-2");
assert_eq!(notes.channel(cx).unwrap().name, "channel-2");
});
// Client B is taken to the notes for channel 2.
@ -759,7 +757,7 @@ async fn test_following_to_channel_notes_without_a_shared_project(
.expect("active item is not a channel view")
});
channel_view_2_b.read_with(cx_b, |notes, cx| {
assert_eq!(notes.channel(cx).name, "channel-2");
assert_eq!(notes.channel(cx).unwrap().name, "channel-2");
});
}

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

File diff suppressed because it is too large Load diff

17
crates/collab/src/tests/random_channel_buffer_tests.rs
View file

@ -48,7 +48,7 @@ impl RandomizedTest for RandomChannelBufferTest {
let db = &server.app_state.db;
for ix in 0..CHANNEL_COUNT {
let id = db
.create_channel(&format!("channel-{ix}"), None, users[0].user_id)
.create_root_channel(&format!("channel-{ix}"), users[0].user_id)
.await
.unwrap();
for user in &users[1..] {
@ -98,15 +98,16 @@ impl RandomizedTest for RandomChannelBufferTest {
30..=40 => {
if let Some(buffer) = channel_buffers.iter().choose(rng) {
let channel_name = buffer.read_with(cx, |b, _| b.channel().name.clone());
let channel_name =
buffer.read_with(cx, |b, cx| b.channel(cx).unwrap().name.clone());
break ChannelBufferOperation::LeaveChannelNotes { channel_name };
}
}
_ => {
if let Some(buffer) = channel_buffers.iter().choose(rng) {
break buffer.read_with(cx, |b, _| {
let channel_name = b.channel().name.clone();
break buffer.read_with(cx, |b, cx| {
let channel_name = b.channel(cx).unwrap().name.clone();
let edits = b
.buffer()
.read_with(cx, |buffer, _| buffer.get_random_edits(rng, 3));
@ -153,7 +154,7 @@ impl RandomizedTest for RandomChannelBufferTest {
let buffer = cx.update(|cx| {
let mut left_buffer = Err(TestError::Inapplicable);
client.channel_buffers().retain(|buffer| {
if buffer.read(cx).channel().name == channel_name {
if buffer.read(cx).channel(cx).unwrap().name == channel_name {
left_buffer = Ok(buffer.clone());
false
} else {
@ -179,7 +180,9 @@ impl RandomizedTest for RandomChannelBufferTest {
client
.channel_buffers()
.iter()
.find(|buffer| buffer.read(cx).channel().name == channel_name)
.find(|buffer| {
buffer.read(cx).channel(cx).unwrap().name == channel_name
})
.cloned()
})
.ok_or_else(|| TestError::Inapplicable)?;
@ -250,7 +253,7 @@ impl RandomizedTest for RandomChannelBufferTest {
if let Some(channel_buffer) = client
.channel_buffers()
.iter()
.find(|b| b.read(cx).channel().id == channel_id.to_proto())
.find(|b| b.read(cx).channel_id == channel_id.to_proto())
{
let channel_buffer = channel_buffer.read(cx);

32
crates/collab/src/tests/test_server.rs
View file

@ -611,38 +611,6 @@ impl TestClient {
) -> WindowHandle<Workspace> {
cx.add_window(|cx| Workspace::new(0, project.clone(), self.app_state.clone(), cx))
}
pub async fn add_admin_to_channel(
&self,
user: (&TestClient, &mut TestAppContext),
channel: u64,
cx_self: &mut TestAppContext,
) {
let (other_client, other_cx) = user;
cx_self
.read(ChannelStore::global)
.update(cx_self, |channel_store, cx| {
channel_store.invite_member(
channel,
other_client.user_id().unwrap(),
ChannelRole::Admin,
cx,
)
})
.await
.unwrap();
cx_self.foreground().run_until_parked();
other_cx
.read(ChannelStore::global)
.update(other_cx, |channel_store, cx| {
channel_store.respond_to_channel_invite(channel, true, cx)
})
.await
.unwrap();
}
}
impl Drop for TestClient {