Start work on private projects

crates/collab/src/integration_tests.rs

@@ -504,6 +504,70 @@ async fn test_cancel_join_request(
     );
 }
 
+#[gpui::test(iterations = 3)]
+async fn test_private_projects(
+    deterministic: Arc<Deterministic>,
+    cx_a: &mut TestAppContext,
+    cx_b: &mut TestAppContext,
+) {
+    cx_a.foreground().forbid_parking();
+    let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
+    let mut client_a = server.create_client(cx_a, "user_a").await;
+    let client_b = server.create_client(cx_b, "user_b").await;
+    server
+        .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
+        .await;
+
+    let user_a = UserId::from_proto(client_a.user_id().unwrap());
+
+    let fs = FakeFs::new(cx_a.background());
+    fs.insert_tree("/a", json!({})).await;
+
+    // Create a private project
+    let project_a = cx_a.update(|cx| {
+        Project::local(
+            false,
+            client_a.client.clone(),
+            client_a.user_store.clone(),
+            client_a.language_registry.clone(),
+            fs.clone(),
+            cx,
+        )
+    });
+    client_a.project = Some(project_a.clone());
+
+    // Private projects are not registered with the server.
+    deterministic.run_until_parked();
+    assert!(project_a.read_with(cx_a, |project, _| project.remote_id().is_none()));
+    assert!(client_b
+        .user_store
+        .read_with(cx_b, |store, _| { store.contacts()[0].projects.is_empty() }));
+
+    // The project is registered when it is made public.
+    project_a.update(cx_a, |project, _| project.set_public(true));
+    deterministic.run_until_parked();
+    assert!(project_a.read_with(cx_a, |project, _| project.remote_id().is_some()));
+    assert!(!client_b
+        .user_store
+        .read_with(cx_b, |store, _| { store.contacts()[0].projects.is_empty() }));
+
+    // The project is registered again when it loses and regains connection.
+    server.disconnect_client(user_a);
+    server.forbid_connections();
+    cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
+    // deterministic.run_until_parked();
+    assert!(project_a.read_with(cx_a, |project, _| project.remote_id().is_none()));
+    assert!(client_b
+        .user_store
+        .read_with(cx_b, |store, _| { store.contacts()[0].projects.is_empty() }));
+    server.allow_connections();
+    cx_b.foreground().advance_clock(Duration::from_secs(10));
+    assert!(project_a.read_with(cx_a, |project, _| project.remote_id().is_some()));
+    assert!(!client_b
+        .user_store
+        .read_with(cx_b, |store, _| { store.contacts()[0].projects.is_empty() }));
+}
+
 #[gpui::test(iterations = 10)]
 async fn test_propagate_saves_and_fs_changes(
     cx_a: &mut TestAppContext,
@@ -4009,6 +4073,7 @@ async fn test_random_collaboration(
     let host = server.create_client(&mut host_cx, "host").await;
     let host_project = host_cx.update(|cx| {
         Project::local(
+            true,
             host.client.clone(),
             host.user_store.clone(),
             host_language_registry.clone(),
@@ -4735,6 +4800,7 @@ impl TestClient {
     ) -> (ModelHandle<Project>, WorktreeId) {
         let project = cx.update(|cx| {
             Project::local(
+                true,
                 self.client.clone(),
                 self.user_store.clone(),
                 self.language_registry.clone(),

crates/project/src/project.rs

@@ -8,7 +8,7 @@ use anyhow::{anyhow, Context, Result};
 use client::{proto, Client, PeerId, TypedEnvelope, User, UserStore};
 use clock::ReplicaId;
 use collections::{hash_map, BTreeMap, HashMap, HashSet};
-use futures::{future::Shared, Future, FutureExt, StreamExt, TryFutureExt};
+use futures::{future::Shared, select_biased, Future, FutureExt, StreamExt, TryFutureExt};
 use fuzzy::{PathMatch, PathMatchCandidate, PathMatchCandidateSet};
 use gpui::{
     AnyModelHandle, AppContext, AsyncAppContext, Entity, ModelContext, ModelHandle,
@@ -120,6 +120,7 @@ enum ProjectClientState {
         is_shared: bool,
         remote_id_tx: watch::Sender<Option<u64>>,
         remote_id_rx: watch::Receiver<Option<u64>>,
+        public_tx: watch::Sender<bool>,
         _maintain_remote_id_task: Task<Option<()>>,
     },
     Remote {
@@ -305,6 +306,7 @@ impl Project {
     }
 
     pub fn local(
+        public: bool,
         client: Arc<Client>,
         user_store: ModelHandle<UserStore>,
         languages: Arc<LanguageRegistry>,
@@ -312,24 +314,25 @@ impl Project {
         cx: &mut MutableAppContext,
     ) -> ModelHandle<Self> {
         cx.add_model(|cx: &mut ModelContext<Self>| {
+            let (public_tx, mut public_rx) = watch::channel_with(public);
             let (remote_id_tx, remote_id_rx) = watch::channel();
             let _maintain_remote_id_task = cx.spawn_weak({
-                let rpc = client.clone();
-                move |this, mut cx| {
-                    async move {
-                        let mut status = rpc.status();
-                        while let Some(status) = status.next().await {
-                            if let Some(this) = this.upgrade(&cx) {
-                                if status.is_connected() {
-                                    this.update(&mut cx, |this, cx| this.register(cx)).await?;
-                                } else {
-                                    this.update(&mut cx, |this, cx| this.unregister(cx));
-                                }
-                            }
+                let mut status_rx = client.clone().status();
+                move |this, mut cx| async move {
+                    loop {
+                        select_biased! {
+                            value = status_rx.next().fuse() => { value?; }
+                            value = public_rx.next().fuse() => { value?; }
+                        };
+                        let this = this.upgrade(&cx)?;
+                        if status_rx.borrow().is_connected() && *public_rx.borrow() {
+                            this.update(&mut cx, |this, cx| this.register(cx))
+                                .await
+                                .log_err()?;
+                        } else {
+                            this.update(&mut cx, |this, cx| this.unregister(cx));
                         }
-                        Ok(())
                     }
-                    .log_err()
                 }
             });
 
@@ -346,6 +349,7 @@ impl Project {
                     is_shared: false,
                     remote_id_tx,
                     remote_id_rx,
+                    public_tx,
                     _maintain_remote_id_task,
                 },
                 opened_buffer: (Rc::new(RefCell::new(opened_buffer_tx)), opened_buffer_rx),
@@ -509,7 +513,7 @@ impl Project {
         let http_client = client::test::FakeHttpClient::with_404_response();
         let client = client::Client::new(http_client.clone());
         let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http_client, cx));
-        let project = cx.update(|cx| Project::local(client, user_store, languages, fs, cx));
+        let project = cx.update(|cx| Project::local(true, client, user_store, languages, fs, cx));
         for path in root_paths {
             let (tree, _) = project
                 .update(cx, |project, cx| {
@@ -598,6 +602,20 @@ impl Project {
         &self.fs
     }
 
+    pub fn set_public(&mut self, is_public: bool) {
+        if let ProjectClientState::Local { public_tx, .. } = &mut self.client_state {
+            *public_tx.borrow_mut() = is_public;
+        }
+    }
+
+    pub fn is_public(&mut self) -> bool {
+        if let ProjectClientState::Local { public_tx, .. } = &mut self.client_state {
+            *public_tx.borrow()
+        } else {
+            true
+        }
+    }
+
     fn unregister(&mut self, cx: &mut ModelContext<Self>) {
         self.unshared(cx);
         for worktree in &self.worktrees {
@@ -616,7 +634,11 @@ impl Project {
     }
 
     fn register(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
-        self.unregister(cx);
+        if let ProjectClientState::Local { remote_id_rx, .. } = &self.client_state {
+            if remote_id_rx.borrow().is_some() {
+                return Task::ready(Ok(()));
+            }
+        }
 
         let response = self.client.request(proto::RegisterProject {});
         cx.spawn(|this, mut cx| async move {

crates/workspace/src/workspace.rs

@@ -2407,6 +2407,7 @@ pub fn open_paths(
             cx.add_window((app_state.build_window_options)(), |cx| {
                 let mut workspace = Workspace::new(
                     Project::local(
+                        false,
                         app_state.client.clone(),
                         app_state.user_store.clone(),
                         app_state.languages.clone(),
@@ -2463,6 +2464,7 @@ fn open_new(app_state: &Arc<AppState>, cx: &mut MutableAppContext) {
     let (window_id, workspace) = cx.add_window((app_state.build_window_options)(), |cx| {
         let mut workspace = Workspace::new(
             Project::local(
+                false,
                 app_state.client.clone(),
                 app_state.user_store.clone(),
                 app_state.languages.clone(),

crates/zed/src/zed.rs

@@ -295,6 +295,7 @@ fn open_config_file(
                     let (_, workspace) = cx.add_window((app_state.build_window_options)(), |cx| {
                         let mut workspace = Workspace::new(
                             Project::local(
+                                false,
                                 app_state.client.clone(),
                                 app_state.user_store.clone(),
                                 app_state.languages.clone(),