1use crate::{
2 participant::{LocalParticipant, ParticipantLocation, RemoteParticipant, RemoteVideoTrack},
3 IncomingCall,
4};
5use anyhow::{anyhow, Result};
6use client::{proto, Client, PeerId, TypedEnvelope, User, UserStore};
7use collections::{BTreeMap, HashSet};
8use futures::StreamExt;
9use gpui::{AsyncAppContext, Entity, ModelContext, ModelHandle, MutableAppContext, Task};
10use live_kit_client::{LocalTrackPublication, LocalVideoTrack, RemoteVideoTrackUpdate};
11use postage::stream::Stream;
12use project::Project;
13use std::{mem, sync::Arc};
14use util::{post_inc, ResultExt};
15
16#[derive(Clone, Debug, PartialEq, Eq)]
17pub enum Event {
18 ParticipantLocationChanged {
19 participant_id: PeerId,
20 },
21 RemoteVideoTracksChanged {
22 participant_id: PeerId,
23 },
24 RemoteProjectShared {
25 owner: Arc<User>,
26 project_id: u64,
27 worktree_root_names: Vec<String>,
28 },
29 RemoteProjectUnshared {
30 project_id: u64,
31 },
32 Left,
33}
34
35pub struct Room {
36 id: u64,
37 version: u64,
38 live_kit: Option<LiveKitRoom>,
39 status: RoomStatus,
40 local_participant: LocalParticipant,
41 remote_participants: BTreeMap<PeerId, RemoteParticipant>,
42 pending_participants: Vec<Arc<User>>,
43 participant_user_ids: HashSet<u64>,
44 pending_call_count: usize,
45 leave_when_empty: bool,
46 client: Arc<Client>,
47 user_store: ModelHandle<UserStore>,
48 subscriptions: Vec<client::Subscription>,
49 pending_room_update: Option<Task<()>>,
50}
51
52impl Entity for Room {
53 type Event = Event;
54
55 fn release(&mut self, _: &mut MutableAppContext) {
56 if self.status.is_online() {
57 self.client.send(proto::LeaveRoom {}).log_err();
58 }
59 }
60}
61
62impl Room {
63 fn new(
64 id: u64,
65 version: u64,
66 live_kit_connection_info: Option<proto::LiveKitConnectionInfo>,
67 client: Arc<Client>,
68 user_store: ModelHandle<UserStore>,
69 cx: &mut ModelContext<Self>,
70 ) -> Self {
71 let mut client_status = client.status();
72 cx.spawn_weak(|this, mut cx| async move {
73 let is_connected = client_status
74 .next()
75 .await
76 .map_or(false, |s| s.is_connected());
77 // Even if we're initially connected, any future change of the status means we momentarily disconnected.
78 if !is_connected || client_status.next().await.is_some() {
79 if let Some(this) = this.upgrade(&cx) {
80 let _ = this.update(&mut cx, |this, cx| this.leave(cx));
81 }
82 }
83 })
84 .detach();
85
86 let live_kit_room = if let Some(connection_info) = live_kit_connection_info {
87 let room = live_kit_client::Room::new();
88 let mut status = room.status();
89 // Consume the initial status of the room.
90 let _ = status.try_recv();
91 let _maintain_room = cx.spawn_weak(|this, mut cx| async move {
92 while let Some(status) = status.next().await {
93 let this = if let Some(this) = this.upgrade(&cx) {
94 this
95 } else {
96 break;
97 };
98
99 if status == live_kit_client::ConnectionState::Disconnected {
100 this.update(&mut cx, |this, cx| this.leave(cx).log_err());
101 break;
102 }
103 }
104 });
105
106 let mut track_changes = room.remote_video_track_updates();
107 let _maintain_tracks = cx.spawn_weak(|this, mut cx| async move {
108 while let Some(track_change) = track_changes.next().await {
109 let this = if let Some(this) = this.upgrade(&cx) {
110 this
111 } else {
112 break;
113 };
114
115 this.update(&mut cx, |this, cx| {
116 this.remote_video_track_updated(track_change, cx).log_err()
117 });
118 }
119 });
120
121 cx.foreground()
122 .spawn(room.connect(&connection_info.server_url, &connection_info.token))
123 .detach_and_log_err(cx);
124
125 Some(LiveKitRoom {
126 room,
127 screen_track: ScreenTrack::None,
128 next_publish_id: 0,
129 _maintain_room,
130 _maintain_tracks,
131 })
132 } else {
133 None
134 };
135
136 Self {
137 id,
138 version,
139 live_kit: live_kit_room,
140 status: RoomStatus::Online,
141 participant_user_ids: Default::default(),
142 local_participant: Default::default(),
143 remote_participants: Default::default(),
144 pending_participants: Default::default(),
145 pending_call_count: 0,
146 subscriptions: vec![client.add_message_handler(cx.handle(), Self::handle_room_updated)],
147 leave_when_empty: false,
148 pending_room_update: None,
149 client,
150 user_store,
151 }
152 }
153
154 pub(crate) fn create(
155 called_user_id: u64,
156 initial_project: Option<ModelHandle<Project>>,
157 client: Arc<Client>,
158 user_store: ModelHandle<UserStore>,
159 cx: &mut MutableAppContext,
160 ) -> Task<Result<ModelHandle<Self>>> {
161 cx.spawn(|mut cx| async move {
162 let response = client.request(proto::CreateRoom {}).await?;
163 let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
164 let room = cx.add_model(|cx| {
165 Self::new(
166 room_proto.id,
167 room_proto.version,
168 response.live_kit_connection_info,
169 client,
170 user_store,
171 cx,
172 )
173 });
174
175 let initial_project_id = if let Some(initial_project) = initial_project {
176 let initial_project_id = room
177 .update(&mut cx, |room, cx| {
178 room.share_project(initial_project.clone(), cx)
179 })
180 .await?;
181 Some(initial_project_id)
182 } else {
183 None
184 };
185
186 match room
187 .update(&mut cx, |room, cx| {
188 room.leave_when_empty = true;
189 room.call(called_user_id, initial_project_id, cx)
190 })
191 .await
192 {
193 Ok(()) => Ok(room),
194 Err(error) => Err(anyhow!("room creation failed: {:?}", error)),
195 }
196 })
197 }
198
199 pub(crate) fn join(
200 call: &IncomingCall,
201 client: Arc<Client>,
202 user_store: ModelHandle<UserStore>,
203 cx: &mut MutableAppContext,
204 ) -> Task<Result<ModelHandle<Self>>> {
205 let room_id = call.room_id;
206 cx.spawn(|mut cx| async move {
207 let response = client.request(proto::JoinRoom { id: room_id }).await?;
208 let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
209 let room = cx.add_model(|cx| {
210 Self::new(
211 room_id,
212 0,
213 response.live_kit_connection_info,
214 client,
215 user_store,
216 cx,
217 )
218 });
219 room.update(&mut cx, |room, cx| {
220 room.leave_when_empty = true;
221 room.apply_room_update(room_proto, cx)?;
222 anyhow::Ok(())
223 })?;
224 Ok(room)
225 })
226 }
227
228 fn should_leave(&self) -> bool {
229 self.leave_when_empty
230 && self.pending_room_update.is_none()
231 && self.pending_participants.is_empty()
232 && self.remote_participants.is_empty()
233 && self.pending_call_count == 0
234 }
235
236 pub(crate) fn leave(&mut self, cx: &mut ModelContext<Self>) -> Result<()> {
237 if self.status.is_offline() {
238 return Err(anyhow!("room is offline"));
239 }
240
241 cx.notify();
242 cx.emit(Event::Left);
243 self.status = RoomStatus::Offline;
244 self.remote_participants.clear();
245 self.pending_participants.clear();
246 self.participant_user_ids.clear();
247 self.subscriptions.clear();
248 self.live_kit.take();
249 self.client.send(proto::LeaveRoom {})?;
250 Ok(())
251 }
252
253 pub fn id(&self) -> u64 {
254 self.id
255 }
256
257 pub fn status(&self) -> RoomStatus {
258 self.status
259 }
260
261 pub fn local_participant(&self) -> &LocalParticipant {
262 &self.local_participant
263 }
264
265 pub fn remote_participants(&self) -> &BTreeMap<PeerId, RemoteParticipant> {
266 &self.remote_participants
267 }
268
269 pub fn pending_participants(&self) -> &[Arc<User>] {
270 &self.pending_participants
271 }
272
273 pub fn contains_participant(&self, user_id: u64) -> bool {
274 self.participant_user_ids.contains(&user_id)
275 }
276
277 async fn handle_room_updated(
278 this: ModelHandle<Self>,
279 envelope: TypedEnvelope<proto::RoomUpdated>,
280 _: Arc<Client>,
281 mut cx: AsyncAppContext,
282 ) -> Result<()> {
283 let room = envelope
284 .payload
285 .room
286 .ok_or_else(|| anyhow!("invalid room"))?;
287 this.update(&mut cx, |this, cx| this.apply_room_update(room, cx))
288 }
289
290 fn apply_room_update(
291 &mut self,
292 mut room: proto::Room,
293 cx: &mut ModelContext<Self>,
294 ) -> Result<()> {
295 // Filter ourselves out from the room's participants.
296 let local_participant_ix = room
297 .participants
298 .iter()
299 .position(|participant| Some(participant.user_id) == self.client.user_id());
300 let local_participant = local_participant_ix.map(|ix| room.participants.swap_remove(ix));
301
302 let pending_participant_user_ids = room
303 .pending_participants
304 .iter()
305 .map(|p| p.user_id)
306 .collect::<Vec<_>>();
307 let remote_participant_user_ids = room
308 .participants
309 .iter()
310 .map(|p| p.user_id)
311 .collect::<Vec<_>>();
312 let (remote_participants, pending_participants) =
313 self.user_store.update(cx, move |user_store, cx| {
314 (
315 user_store.get_users(remote_participant_user_ids, cx),
316 user_store.get_users(pending_participant_user_ids, cx),
317 )
318 });
319 self.pending_room_update = Some(cx.spawn(|this, mut cx| async move {
320 let (remote_participants, pending_participants) =
321 futures::join!(remote_participants, pending_participants);
322
323 this.update(&mut cx, |this, cx| {
324 if this.version >= room.version {
325 return;
326 }
327
328 this.participant_user_ids.clear();
329
330 if let Some(participant) = local_participant {
331 this.local_participant.projects = participant.projects;
332 } else {
333 this.local_participant.projects.clear();
334 }
335
336 if let Some(participants) = remote_participants.log_err() {
337 for (participant, user) in room.participants.into_iter().zip(participants) {
338 let peer_id = PeerId(participant.peer_id);
339 this.participant_user_ids.insert(participant.user_id);
340
341 let old_projects = this
342 .remote_participants
343 .get(&peer_id)
344 .into_iter()
345 .flat_map(|existing| &existing.projects)
346 .map(|project| project.id)
347 .collect::<HashSet<_>>();
348 let new_projects = participant
349 .projects
350 .iter()
351 .map(|project| project.id)
352 .collect::<HashSet<_>>();
353
354 for project in &participant.projects {
355 if !old_projects.contains(&project.id) {
356 cx.emit(Event::RemoteProjectShared {
357 owner: user.clone(),
358 project_id: project.id,
359 worktree_root_names: project.worktree_root_names.clone(),
360 });
361 }
362 }
363
364 for unshared_project_id in old_projects.difference(&new_projects) {
365 cx.emit(Event::RemoteProjectUnshared {
366 project_id: *unshared_project_id,
367 });
368 }
369
370 let location = ParticipantLocation::from_proto(participant.location)
371 .unwrap_or(ParticipantLocation::External);
372 if let Some(remote_participant) = this.remote_participants.get_mut(&peer_id)
373 {
374 remote_participant.projects = participant.projects;
375 if location != remote_participant.location {
376 remote_participant.location = location;
377 cx.emit(Event::ParticipantLocationChanged {
378 participant_id: peer_id,
379 });
380 }
381 } else {
382 this.remote_participants.insert(
383 peer_id,
384 RemoteParticipant {
385 user: user.clone(),
386 projects: participant.projects,
387 location,
388 tracks: Default::default(),
389 },
390 );
391
392 if let Some(live_kit) = this.live_kit.as_ref() {
393 let tracks =
394 live_kit.room.remote_video_tracks(&peer_id.0.to_string());
395 for track in tracks {
396 this.remote_video_track_updated(
397 RemoteVideoTrackUpdate::Subscribed(track),
398 cx,
399 )
400 .log_err();
401 }
402 }
403 }
404 }
405
406 this.remote_participants.retain(|_, participant| {
407 if this.participant_user_ids.contains(&participant.user.id) {
408 true
409 } else {
410 for project in &participant.projects {
411 cx.emit(Event::RemoteProjectUnshared {
412 project_id: project.id,
413 });
414 }
415 false
416 }
417 });
418 }
419
420 if let Some(pending_participants) = pending_participants.log_err() {
421 this.pending_participants = pending_participants;
422 for participant in &this.pending_participants {
423 this.participant_user_ids.insert(participant.id);
424 }
425 }
426
427 this.pending_room_update.take();
428 if this.should_leave() {
429 let _ = this.leave(cx);
430 }
431
432 this.version = room.version;
433 this.check_invariants();
434 cx.notify();
435 });
436 }));
437
438 cx.notify();
439 Ok(())
440 }
441
442 fn remote_video_track_updated(
443 &mut self,
444 change: RemoteVideoTrackUpdate,
445 cx: &mut ModelContext<Self>,
446 ) -> Result<()> {
447 match change {
448 RemoteVideoTrackUpdate::Subscribed(track) => {
449 let peer_id = PeerId(track.publisher_id().parse()?);
450 let track_id = track.sid().to_string();
451 let participant = self
452 .remote_participants
453 .get_mut(&peer_id)
454 .ok_or_else(|| anyhow!("subscribed to track by unknown participant"))?;
455 participant.tracks.insert(
456 track_id.clone(),
457 Arc::new(RemoteVideoTrack {
458 live_kit_track: track,
459 }),
460 );
461 cx.emit(Event::RemoteVideoTracksChanged {
462 participant_id: peer_id,
463 });
464 }
465 RemoteVideoTrackUpdate::Unsubscribed {
466 publisher_id,
467 track_id,
468 } => {
469 let peer_id = PeerId(publisher_id.parse()?);
470 let participant = self
471 .remote_participants
472 .get_mut(&peer_id)
473 .ok_or_else(|| anyhow!("unsubscribed from track by unknown participant"))?;
474 participant.tracks.remove(&track_id);
475 cx.emit(Event::RemoteVideoTracksChanged {
476 participant_id: peer_id,
477 });
478 }
479 }
480
481 cx.notify();
482 Ok(())
483 }
484
485 fn check_invariants(&self) {
486 #[cfg(any(test, feature = "test-support"))]
487 {
488 for participant in self.remote_participants.values() {
489 assert!(self.participant_user_ids.contains(&participant.user.id));
490 }
491
492 for participant in &self.pending_participants {
493 assert!(self.participant_user_ids.contains(&participant.id));
494 }
495
496 assert_eq!(
497 self.participant_user_ids.len(),
498 self.remote_participants.len() + self.pending_participants.len()
499 );
500 }
501 }
502
503 pub(crate) fn call(
504 &mut self,
505 called_user_id: u64,
506 initial_project_id: Option<u64>,
507 cx: &mut ModelContext<Self>,
508 ) -> Task<Result<()>> {
509 if self.status.is_offline() {
510 return Task::ready(Err(anyhow!("room is offline")));
511 }
512
513 cx.notify();
514 let client = self.client.clone();
515 let room_id = self.id;
516 self.pending_call_count += 1;
517 cx.spawn(|this, mut cx| async move {
518 let result = client
519 .request(proto::Call {
520 room_id,
521 called_user_id,
522 initial_project_id,
523 })
524 .await;
525 this.update(&mut cx, |this, cx| {
526 this.pending_call_count -= 1;
527 if this.should_leave() {
528 this.leave(cx)?;
529 }
530 result
531 })?;
532 Ok(())
533 })
534 }
535
536 pub(crate) fn share_project(
537 &mut self,
538 project: ModelHandle<Project>,
539 cx: &mut ModelContext<Self>,
540 ) -> Task<Result<u64>> {
541 if let Some(project_id) = project.read(cx).remote_id() {
542 return Task::ready(Ok(project_id));
543 }
544
545 let request = self.client.request(proto::ShareProject {
546 room_id: self.id(),
547 worktrees: project
548 .read(cx)
549 .worktrees(cx)
550 .map(|worktree| {
551 let worktree = worktree.read(cx);
552 proto::WorktreeMetadata {
553 id: worktree.id().to_proto(),
554 root_name: worktree.root_name().into(),
555 visible: worktree.is_visible(),
556 abs_path: worktree.abs_path().to_string_lossy().into(),
557 }
558 })
559 .collect(),
560 });
561 cx.spawn(|this, mut cx| async move {
562 let response = request.await?;
563
564 project.update(&mut cx, |project, cx| {
565 project
566 .shared(response.project_id, cx)
567 .detach_and_log_err(cx)
568 });
569
570 // If the user's location is in this project, it changes from UnsharedProject to SharedProject.
571 this.update(&mut cx, |this, cx| {
572 let active_project = this.local_participant.active_project.as_ref();
573 if active_project.map_or(false, |location| *location == project) {
574 this.set_location(Some(&project), cx)
575 } else {
576 Task::ready(Ok(()))
577 }
578 })
579 .await?;
580
581 Ok(response.project_id)
582 })
583 }
584
585 pub(crate) fn set_location(
586 &mut self,
587 project: Option<&ModelHandle<Project>>,
588 cx: &mut ModelContext<Self>,
589 ) -> Task<Result<()>> {
590 if self.status.is_offline() {
591 return Task::ready(Err(anyhow!("room is offline")));
592 }
593
594 let client = self.client.clone();
595 let room_id = self.id;
596 let location = if let Some(project) = project {
597 self.local_participant.active_project = Some(project.downgrade());
598 if let Some(project_id) = project.read(cx).remote_id() {
599 proto::participant_location::Variant::SharedProject(
600 proto::participant_location::SharedProject { id: project_id },
601 )
602 } else {
603 proto::participant_location::Variant::UnsharedProject(
604 proto::participant_location::UnsharedProject {},
605 )
606 }
607 } else {
608 self.local_participant.active_project = None;
609 proto::participant_location::Variant::External(proto::participant_location::External {})
610 };
611
612 cx.notify();
613 cx.foreground().spawn(async move {
614 client
615 .request(proto::UpdateParticipantLocation {
616 room_id,
617 location: Some(proto::ParticipantLocation {
618 variant: Some(location),
619 }),
620 })
621 .await?;
622 Ok(())
623 })
624 }
625
626 pub fn is_screen_sharing(&self) -> bool {
627 self.live_kit.as_ref().map_or(false, |live_kit| {
628 !matches!(live_kit.screen_track, ScreenTrack::None)
629 })
630 }
631
632 pub fn share_screen(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
633 if self.status.is_offline() {
634 return Task::ready(Err(anyhow!("room is offline")));
635 } else if self.is_screen_sharing() {
636 return Task::ready(Err(anyhow!("screen was already shared")));
637 }
638
639 let (displays, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
640 let publish_id = post_inc(&mut live_kit.next_publish_id);
641 live_kit.screen_track = ScreenTrack::Pending { publish_id };
642 cx.notify();
643 (live_kit.room.display_sources(), publish_id)
644 } else {
645 return Task::ready(Err(anyhow!("live-kit was not initialized")));
646 };
647
648 cx.spawn_weak(|this, mut cx| async move {
649 let publish_track = async {
650 let displays = displays.await?;
651 let display = displays
652 .first()
653 .ok_or_else(|| anyhow!("no display found"))?;
654 let track = LocalVideoTrack::screen_share_for_display(&display);
655 this.upgrade(&cx)
656 .ok_or_else(|| anyhow!("room was dropped"))?
657 .read_with(&cx, |this, _| {
658 this.live_kit
659 .as_ref()
660 .map(|live_kit| live_kit.room.publish_video_track(&track))
661 })
662 .ok_or_else(|| anyhow!("live-kit was not initialized"))?
663 .await
664 };
665
666 let publication = publish_track.await;
667 this.upgrade(&cx)
668 .ok_or_else(|| anyhow!("room was dropped"))?
669 .update(&mut cx, |this, cx| {
670 let live_kit = this
671 .live_kit
672 .as_mut()
673 .ok_or_else(|| anyhow!("live-kit was not initialized"))?;
674
675 let canceled = if let ScreenTrack::Pending {
676 publish_id: cur_publish_id,
677 } = &live_kit.screen_track
678 {
679 *cur_publish_id != publish_id
680 } else {
681 true
682 };
683
684 match publication {
685 Ok(publication) => {
686 if canceled {
687 live_kit.room.unpublish_track(publication);
688 } else {
689 live_kit.screen_track = ScreenTrack::Published(publication);
690 cx.notify();
691 }
692 Ok(())
693 }
694 Err(error) => {
695 if canceled {
696 Ok(())
697 } else {
698 live_kit.screen_track = ScreenTrack::None;
699 cx.notify();
700 Err(error)
701 }
702 }
703 }
704 })
705 })
706 }
707
708 pub fn unshare_screen(&mut self, cx: &mut ModelContext<Self>) -> Result<()> {
709 if self.status.is_offline() {
710 return Err(anyhow!("room is offline"));
711 }
712
713 let live_kit = self
714 .live_kit
715 .as_mut()
716 .ok_or_else(|| anyhow!("live-kit was not initialized"))?;
717 match mem::take(&mut live_kit.screen_track) {
718 ScreenTrack::None => Err(anyhow!("screen was not shared")),
719 ScreenTrack::Pending { .. } => {
720 cx.notify();
721 Ok(())
722 }
723 ScreenTrack::Published(track) => {
724 live_kit.room.unpublish_track(track);
725 cx.notify();
726 Ok(())
727 }
728 }
729 }
730
731 #[cfg(any(test, feature = "test-support"))]
732 pub fn set_display_sources(&self, sources: Vec<live_kit_client::MacOSDisplay>) {
733 self.live_kit
734 .as_ref()
735 .unwrap()
736 .room
737 .set_display_sources(sources);
738 }
739}
740
741struct LiveKitRoom {
742 room: Arc<live_kit_client::Room>,
743 screen_track: ScreenTrack,
744 next_publish_id: usize,
745 _maintain_room: Task<()>,
746 _maintain_tracks: Task<()>,
747}
748
749enum ScreenTrack {
750 None,
751 Pending { publish_id: usize },
752 Published(LocalTrackPublication),
753}
754
755impl Default for ScreenTrack {
756 fn default() -> Self {
757 Self::None
758 }
759}
760
761#[derive(Copy, Clone, PartialEq, Eq)]
762pub enum RoomStatus {
763 Online,
764 Offline,
765}
766
767impl RoomStatus {
768 pub fn is_offline(&self) -> bool {
769 matches!(self, RoomStatus::Offline)
770 }
771
772 pub fn is_online(&self) -> bool {
773 matches!(self, RoomStatus::Online)
774 }
775}