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

single click channel (#7596)

Browse source 
- Open channel notes and chat on channel click
- WIP
- Fix compile error
- Don't join live kit until requested
- Track in_call state separately from in_room



Release Notes:

- Improved channels: you can now be in a channel without joining the
audio call automatically

**or**

- N/A

---------

Co-authored-by: Nathan Sobo <nathan@zed.dev>
This commit is contained in:
Conrad Irwin 2024-02-09 14:18:27 -07:00 • committed by GitHub
parent 2b39a9512a
commit efe23ebfcd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 659 additions and 489 deletions
Download patch file Download diff file Expand all files Collapse all files

3
crates/collab/migrations.sqlite/20221109000000_test_schema.sql
View file

@ -163,7 +163,8 @@ CREATE TABLE "room_participants" (
"calling_connection_id" INTEGER NOT NULL,
"calling_connection_server_id" INTEGER REFERENCES servers (id) ON DELETE SET NULL,
"participant_index" INTEGER,
"role" TEXT
"role" TEXT,
"in_call" BOOLEAN NOT NULL DEFAULT FALSE
);
CREATE UNIQUE INDEX "index_room_participants_on_user_id" ON "room_participants" ("user_id");
CREATE INDEX "index_room_participants_on_room_id" ON "room_participants" ("room_id");

3
crates/collab/migrations/20240207041417_add_in_call_column_to_room_participants.sql Normal file
View file

@ -0,0 +1,3 @@
-- Add migration script here
ALTER TABLE room_participants ADD COLUMN in_call BOOL NOT NULL DEFAULT FALSE;

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

@ -97,11 +97,57 @@ impl Database {
.await
}
pub async fn set_in_channel_call(
&self,
channel_id: ChannelId,
user_id: UserId,
in_call: bool,
) -> Result<(proto::Room, ChannelRole)> {
self.transaction(move |tx| async move {
let channel = self.get_channel_internal(channel_id, &*tx).await?;
let role = self.channel_role_for_user(&channel, user_id, &*tx).await?;
if role.is_none() || role == Some(ChannelRole::Banned) {
Err(ErrorCode::Forbidden.anyhow())?
}
let role = role.unwrap();
let Some(room) = room::Entity::find()
.filter(room::Column::ChannelId.eq(channel_id))
.one(&*tx)
.await?
else {
Err(anyhow!("no room exists"))?
};
let result = room_participant::Entity::update_many()
.filter(
Condition::all()
.add(room_participant::Column::RoomId.eq(room.id))
.add(room_participant::Column::UserId.eq(user_id)),
)
.set(room_participant::ActiveModel {
in_call: ActiveValue::Set(in_call),
..Default::default()
})
.exec(&*tx)
.await?;
if result.rows_affected != 1 {
Err(anyhow!("not in channel"))?
}
let room = self.get_room(room.id, &*tx).await?;
Ok((room, role))
})
.await
}
/// Adds a user to the specified channel.
pub async fn join_channel(
&self,
channel_id: ChannelId,
user_id: UserId,
autojoin: bool,
connection: ConnectionId,
environment: &str,
) -> Result<(JoinRoom, Option<MembershipUpdated>, ChannelRole)> {
@ -166,7 +212,7 @@ impl Database {
.get_or_create_channel_room(channel_id, &live_kit_room, environment, &*tx)
.await?;
self.join_channel_room_internal(room_id, user_id, connection, role, &*tx)
self.join_channel_room_internal(room_id, user_id, autojoin, connection, role, &*tx)
.await
.map(|jr| (jr, accept_invite_result, role))
})

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

@ -135,6 +135,7 @@ impl Database {
))),
participant_index: ActiveValue::set(Some(0)),
role: ActiveValue::set(Some(ChannelRole::Admin)),
in_call: ActiveValue::set(true),
id: ActiveValue::NotSet,
location_kind: ActiveValue::NotSet,
@ -187,6 +188,7 @@ impl Database {
))),
initial_project_id: ActiveValue::set(initial_project_id),
role: ActiveValue::set(Some(called_user_role)),
in_call: ActiveValue::set(true),
id: ActiveValue::NotSet,
answering_connection_id: ActiveValue::NotSet,
@ -414,6 +416,7 @@ impl Database {
&self,
room_id: RoomId,
user_id: UserId,
autojoin: bool,
connection: ConnectionId,
role: ChannelRole,
tx: &DatabaseTransaction,
@ -437,6 +440,8 @@ impl Database {
))),
participant_index: ActiveValue::Set(Some(participant_index)),
role: ActiveValue::set(Some(role)),
in_call: ActiveValue::set(autojoin),
id: ActiveValue::NotSet,
location_kind: ActiveValue::NotSet,
location_project_id: ActiveValue::NotSet,
@ -1258,6 +1263,7 @@ impl Database {
location: Some(proto::ParticipantLocation { variant: location }),
participant_index: participant_index as u32,
role: db_participant.role.unwrap_or(ChannelRole::Member).into(),
in_call: db_participant.in_call,
},
);
} else {

1
crates/collab/src/db/tables/room_participant.rs
View file

@ -20,6 +20,7 @@ pub struct Model {
pub calling_connection_server_id: Option<ServerId>,
pub participant_index: Option<i32>,
pub role: Option<ChannelRole>,
pub in_call: bool,
}
impl Model {

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

@ -138,6 +138,7 @@ async fn test_joining_channels(db: &Arc<Database>) {
.join_channel(
channel_1,
user_1,
false,
ConnectionId { owner_id, id: 1 },
TEST_RELEASE_CHANNEL,
)
@ -732,9 +733,15 @@ async fn test_guest_access(db: &Arc<Database>) {
.await
.is_err());
db.join_channel(zed_channel, guest, guest_connection, TEST_RELEASE_CHANNEL)
.await
.unwrap();
db.join_channel(
zed_channel,
guest,
false,
guest_connection,
TEST_RELEASE_CHANNEL,
)
.await
.unwrap();
assert!(db
.join_channel_chat(zed_channel, guest_connection, guest)

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

@ -105,6 +105,7 @@ struct Session {
zed_environment: Arc<str>,
user_id: UserId,
connection_id: ConnectionId,
zed_version: SemanticVersion,
db: Arc<tokio::sync::Mutex<DbHandle>>,
peer: Arc<Peer>,
connection_pool: Arc<parking_lot::Mutex<ConnectionPool>>,
@ -131,6 +132,19 @@ impl Session {
_not_send: PhantomData,
}
}
fn endpoint_removed_in(&self, endpoint: &str, version: SemanticVersion) -> anyhow::Result<()> {
if self.zed_version > version {
Err(anyhow!(
"{} was removed in {} (you're on {})",
endpoint,
version,
self.zed_version
))
} else {
Ok(())
}
}
}
impl fmt::Debug for Session {
@ -274,8 +288,11 @@ impl Server {
.add_request_handler(get_channel_members)
.add_request_handler(respond_to_channel_invite)
.add_request_handler(join_channel)
.add_request_handler(join_channel2)
.add_request_handler(join_channel_chat)
.add_message_handler(leave_channel_chat)
.add_request_handler(join_channel_call)
.add_request_handler(leave_channel_call)
.add_request_handler(send_channel_message)
.add_request_handler(remove_channel_message)
.add_request_handler(get_channel_messages)
@ -559,6 +576,7 @@ impl Server {
connection: Connection,
address: String,
user: User,
zed_version: SemanticVersion,
impersonator: Option<User>,
mut send_connection_id: Option<oneshot::Sender<ConnectionId>>,
executor: Executor,
@ -616,6 +634,7 @@ impl Server {
let session = Session {
user_id,
connection_id,
zed_version,
db: Arc::new(tokio::sync::Mutex::new(DbHandle(this.app_state.db.clone()))),
zed_environment: this.app_state.config.zed_environment.clone(),
peer: this.peer.clone(),
@ -866,7 +885,7 @@ pub fn routes(server: Arc<Server>) -> Router<Body> {
pub async fn handle_websocket_request(
TypedHeader(ProtocolVersion(protocol_version)): TypedHeader<ProtocolVersion>,
_app_version_header: Option<TypedHeader<AppVersionHeader>>,
app_version_header: Option<TypedHeader<AppVersionHeader>>,
ConnectInfo(socket_address): ConnectInfo<SocketAddr>,
Extension(server): Extension<Arc<Server>>,
Extension(user): Extension<User>,
@ -881,6 +900,12 @@ pub async fn handle_websocket_request(
.into_response();
}
// zed 0.122.x was the first version that sent an app header, so once that hits stable
// we can return UPGRADE_REQUIRED instead of unwrap_or_default();
let app_version = app_version_header
.map(|header| header.0 .0)
.unwrap_or_default();
let socket_address = socket_address.to_string();
ws.on_upgrade(move |socket| {
use util::ResultExt;
@ -895,6 +920,7 @@ pub async fn handle_websocket_request(
connection,
socket_address,
user,
app_version,
impersonator.0,
None,
Executor::Production,
@ -1037,7 +1063,7 @@ async fn join_room(
let channel_id = session.db().await.channel_id_for_room(room_id).await?;
if let Some(channel_id) = channel_id {
return join_channel_internal(channel_id, Box::new(response), session).await;
return join_channel_internal(channel_id, true, Box::new(response), session).await;
}
let joined_room = {
@ -2700,14 +2726,67 @@ async fn respond_to_channel_invite(
Ok(())
}
/// Join the channels' room
/// Join the channels' call
async fn join_channel(
request: proto::JoinChannel,
response: Response<proto::JoinChannel>,
session: Session,
) -> Result<()> {
session.endpoint_removed_in("join_channel", "0.123.0".parse().unwrap())?;
let channel_id = ChannelId::from_proto(request.channel_id);
join_channel_internal(channel_id, Box::new(response), session).await
join_channel_internal(channel_id, true, Box::new(response), session).await
}
async fn join_channel2(
request: proto::JoinChannel2,
response: Response<proto::JoinChannel2>,
session: Session,
) -> Result<()> {
let channel_id = ChannelId::from_proto(request.channel_id);
join_channel_internal(channel_id, false, Box::new(response), session).await
}
async fn join_channel_call(
request: proto::JoinChannelCall,
response: Response<proto::JoinChannelCall>,
session: Session,
) -> Result<()> {
let channel_id = ChannelId::from_proto(request.channel_id);
let db = session.db().await;
let (joined_room, role) = db
.set_in_channel_call(channel_id, session.user_id, true)
.await?;
let Some(connection_info) = session.live_kit_client.as_ref().and_then(|live_kit| {
live_kit_info_for_user(live_kit, &session.user_id, role, &joined_room.live_kit_room)
}) else {
Err(anyhow!("no live kit token info"))?
};
room_updated(&joined_room, &session.peer);
response.send(proto::JoinChannelCallResponse {
live_kit_connection_info: Some(connection_info),
})?;
Ok(())
}
async fn leave_channel_call(
request: proto::LeaveChannelCall,
response: Response<proto::LeaveChannelCall>,
session: Session,
) -> Result<()> {
let channel_id = ChannelId::from_proto(request.channel_id);
let db = session.db().await;
let (joined_room, _) = db
.set_in_channel_call(channel_id, session.user_id, false)
.await?;
room_updated(&joined_room, &session.peer);
response.send(proto::Ack {})?;
Ok(())
}
trait JoinChannelInternalResponse {
@ -2723,9 +2802,15 @@ impl JoinChannelInternalResponse for Response<proto::JoinRoom> {
Response::<proto::JoinRoom>::send(self, result)
}
}
impl JoinChannelInternalResponse for Response<proto::JoinChannel2> {
fn send(self, result: proto::JoinRoomResponse) -> Result<()> {
Response::<proto::JoinChannel2>::send(self, result)
}
}
async fn join_channel_internal(
channel_id: ChannelId,
autojoin: bool,
response: Box<impl JoinChannelInternalResponse>,
session: Session,
) -> Result<()> {
@ -2737,39 +2822,22 @@ async fn join_channel_internal(
.join_channel(
channel_id,
session.user_id,
autojoin,
session.connection_id,
session.zed_environment.as_ref(),
)
.await?;
let live_kit_connection_info = session.live_kit_client.as_ref().and_then(|live_kit| {
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()?,
)
} 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,
})
if !autojoin {
return None;
}
live_kit_info_for_user(
live_kit,
&session.user_id,
role,
&joined_room.room.live_kit_room,
)
});
response.send(proto::JoinRoomResponse {
@ -2805,6 +2873,35 @@ async fn join_channel_internal(
Ok(())
}
fn live_kit_info_for_user(
live_kit: &Arc<dyn live_kit_server::api::Client>,
user_id: &UserId,
role: ChannelRole,
live_kit_room: &String,
) -> Option<LiveKitConnectionInfo> {
let (can_publish, token) = if role == ChannelRole::Guest {
(
false,
live_kit
.guest_token(live_kit_room, &user_id.to_string())
.trace_err()?,
)
} else {
(
true,
live_kit
.room_token(live_kit_room, &user_id.to_string())
.trace_err()?,
)
};
Some(LiveKitConnectionInfo {
server_url: live_kit.url().into(),
token,
can_publish,
})
}
/// Start editing the channel notes
async fn join_channel_buffer(
request: proto::JoinChannelBuffer,

13
crates/collab/src/tests/channel_guest_tests.rs
View file

@ -1,4 +1,7 @@
use crate::{db::ChannelId, tests::TestServer};
use crate::{
db::ChannelId,
tests::{test_server::join_channel_call, TestServer},
};
use call::ActiveCall;
use editor::Editor;
use gpui::{BackgroundExecutor, TestAppContext};
@ -32,7 +35,7 @@ async fn test_channel_guests(
cx_a.executor().run_until_parked();
// Client B joins channel A as a guest
cx_b.update(|cx| workspace::join_channel(channel_id, client_b.app_state.clone(), None, cx))
cx_b.update(|cx| workspace::open_channel(channel_id, client_b.app_state.clone(), None, cx))
.await
.unwrap();
@ -72,7 +75,7 @@ async fn test_channel_guest_promotion(cx_a: &mut TestAppContext, cx_b: &mut Test
.await;
let project_a = client_a.build_test_project(cx_a).await;
cx_a.update(|cx| workspace::join_channel(channel_id, client_a.app_state.clone(), None, cx))
cx_a.update(|cx| workspace::open_channel(channel_id, client_a.app_state.clone(), None, cx))
.await
.unwrap();
@ -84,11 +87,13 @@ async fn test_channel_guest_promotion(cx_a: &mut TestAppContext, cx_b: &mut Test
cx_a.run_until_parked();
// Client B joins channel A as a guest
cx_b.update(|cx| workspace::join_channel(channel_id, client_b.app_state.clone(), None, cx))
cx_b.update(|cx| workspace::open_channel(channel_id, client_b.app_state.clone(), None, cx))
.await
.unwrap();
cx_a.run_until_parked();
join_channel_call(cx_b).await.unwrap();
// client B opens 1.txt as a guest
let (workspace_b, cx_b) = client_b.active_workspace(cx_b);
let room_b = cx_b

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

@ -1,7 +1,7 @@
use crate::{
db::{self, UserId},
rpc::RECONNECT_TIMEOUT,
tests::{room_participants, RoomParticipants, TestServer},
tests::{room_participants, test_server::join_channel_call, RoomParticipants, TestServer},
};
use call::ActiveCall;
use channel::{ChannelId, ChannelMembership, ChannelStore};
@ -382,6 +382,7 @@ async fn test_channel_room(
.update(cx_a, |active_call, cx| active_call.join_channel(zed_id, cx))
.await
.unwrap();
join_channel_call(cx_a).await.unwrap();
// Give everyone a chance to observe user A joining
executor.run_until_parked();
@ -429,7 +430,7 @@ async fn test_channel_room(
.update(cx_b, |active_call, cx| active_call.join_channel(zed_id, cx))
.await
.unwrap();
join_channel_call(cx_b).await.unwrap();
executor.run_until_parked();
cx_a.read(|cx| {
@ -552,6 +553,9 @@ async fn test_channel_room(
.await
.unwrap();
join_channel_call(cx_a).await.unwrap();
join_channel_call(cx_b).await.unwrap();
executor.run_until_parked();
let room_a =

6
crates/collab/src/tests/following_tests.rs
View file

@ -24,7 +24,7 @@ use workspace::{
use super::TestClient;
#[gpui::test(iterations = 10)]
#[gpui::test]
async fn test_basic_following(
cx_a: &mut TestAppContext,
cx_b: &mut TestAppContext,
@ -437,6 +437,7 @@ async fn test_basic_following(
})
.await
.unwrap();
executor.run_until_parked();
let shared_screen = workspace_a.update(cx_a, |workspace, cx| {
workspace
@ -522,6 +523,7 @@ async fn test_basic_following(
workspace_a.update(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
None
);
executor.run_until_parked();
}
#[gpui::test]
@ -2004,7 +2006,7 @@ async fn join_channel(
client: &TestClient,
cx: &mut TestAppContext,
) -> anyhow::Result<()> {
cx.update(|cx| workspace::join_channel(channel_id, client.app_state.clone(), None, cx))
cx.update(|cx| workspace::open_channel(channel_id, client.app_state.clone(), None, cx))
.await
}

44
crates/collab/src/tests/integration_tests.rs
View file

@ -1881,7 +1881,7 @@ fn active_call_events(cx: &mut TestAppContext) -> Rc<RefCell<Vec<room::Event>>>
}
#[gpui::test]
async fn test_mute_deafen(
async fn test_mute(
executor: BackgroundExecutor,
cx_a: &mut TestAppContext,
cx_b: &mut TestAppContext,
@ -1920,7 +1920,7 @@ async fn test_mute_deafen(
room_a.read_with(cx_a, |room, _| assert!(!room.is_muted()));
room_b.read_with(cx_b, |room, _| assert!(!room.is_muted()));
// Users A and B are both muted.
// Users A and B are both unmuted.
assert_eq!(
participant_audio_state(&room_a, cx_a),
&[ParticipantAudioState {
@ -1962,30 +1962,6 @@ async fn test_mute_deafen(
}]
);
// User A deafens
room_a.update(cx_a, |room, cx| room.toggle_deafen(cx));
executor.run_until_parked();
// User A does not hear user B.
room_a.read_with(cx_a, |room, _| assert!(room.is_muted()));
room_b.read_with(cx_b, |room, _| assert!(!room.is_muted()));
assert_eq!(
participant_audio_state(&room_a, cx_a),
&[ParticipantAudioState {
user_id: client_b.user_id().unwrap(),
is_muted: false,
audio_tracks_playing: vec![false],
}]
);
assert_eq!(
participant_audio_state(&room_b, cx_b),
&[ParticipantAudioState {
user_id: client_a.user_id().unwrap(),
is_muted: true,
audio_tracks_playing: vec![true],
}]
);
// User B calls user C, C joins.
active_call_b
.update(cx_b, |call, cx| {
@ -2000,22 +1976,6 @@ async fn test_mute_deafen(
.unwrap();
executor.run_until_parked();
// User A does not hear users B or C.
assert_eq!(
participant_audio_state(&room_a, cx_a),
&[
ParticipantAudioState {
user_id: client_b.user_id().unwrap(),
is_muted: false,
audio_tracks_playing: vec![false],
},
ParticipantAudioState {
user_id: client_c.user_id().unwrap(),
is_muted: false,
audio_tracks_playing: vec![false],
}
]
);
assert_eq!(
participant_audio_state(&room_b, cx_b),
&[

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

@ -13,7 +13,7 @@ use client::{
use collections::{HashMap, HashSet};
use fs::FakeFs;
use futures::{channel::oneshot, StreamExt as _};
use gpui::{BackgroundExecutor, Context, Model, TestAppContext, View, VisualTestContext};
use gpui::{BackgroundExecutor, Context, Model, Task, TestAppContext, View, VisualTestContext};
use language::LanguageRegistry;
use node_runtime::FakeNodeRuntime;
@ -36,7 +36,7 @@ use std::{
Arc,
},
};
use util::http::FakeHttpClient;
use util::{http::FakeHttpClient, SemanticVersion};
use workspace::{Workspace, WorkspaceStore};
pub struct TestServer {
@ -230,6 +230,7 @@ impl TestServer {
server_conn,
client_name,
user,
SemanticVersion::default(),
None,
Some(connection_id_tx),
Executor::Deterministic(cx.background_executor().clone()),
@ -685,7 +686,7 @@ impl TestClient {
channel_id: u64,
cx: &'a mut TestAppContext,
) -> (View<Workspace>, &'a mut VisualTestContext) {
cx.update(|cx| workspace::join_channel(channel_id, self.app_state.clone(), None, cx))
cx.update(|cx| workspace::open_channel(channel_id, self.app_state.clone(), None, cx))
.await
.unwrap();
cx.run_until_parked();
@ -760,6 +761,11 @@ impl TestClient {
}
}
pub fn join_channel_call(cx: &mut TestAppContext) -> Task<anyhow::Result<()>> {
let room = cx.read(|cx| ActiveCall::global(cx).read(cx).room().cloned());
room.unwrap().update(cx, |room, cx| room.join_call(cx))
}
impl Drop for TestClient {
fn drop(&mut self) {
self.app_state.client.teardown();