Move Store::start_language_server to Db

crates/collab/migrations.sqlite/20221109000000_test_schema.sql

@@ -48,8 +48,8 @@ CREATE TABLE "projects" (
 );
 
 CREATE TABLE "worktrees" (
-    "id" INTEGER NOT NULL,
     "project_id" INTEGER NOT NULL REFERENCES projects (id) ON DELETE CASCADE,
+    "id" INTEGER NOT NULL,
     "root_name" VARCHAR NOT NULL,
     "abs_path" VARCHAR NOT NULL,
     "visible" BOOL NOT NULL,
@@ -60,9 +60,9 @@ CREATE TABLE "worktrees" (
 CREATE INDEX "index_worktrees_on_project_id" ON "worktrees" ("project_id");
 
 CREATE TABLE "worktree_entries" (
-    "id" INTEGER NOT NULL,
     "project_id" INTEGER NOT NULL,
     "worktree_id" INTEGER NOT NULL,
+    "id" INTEGER NOT NULL,
     "is_dir" BOOL NOT NULL,
     "path" VARCHAR NOT NULL,
     "inode" INTEGER NOT NULL,
@@ -76,9 +76,9 @@ CREATE TABLE "worktree_entries" (
 CREATE INDEX "index_worktree_entries_on_project_id_and_worktree_id" ON "worktree_entries" ("project_id", "worktree_id");
 
 CREATE TABLE "worktree_diagnostic_summaries" (
-    "path" VARCHAR NOT NULL,
     "project_id" INTEGER NOT NULL,
     "worktree_id" INTEGER NOT NULL,
+    "path" VARCHAR NOT NULL,
     "language_server_id" INTEGER NOT NULL,
     "error_count" INTEGER NOT NULL,
     "warning_count" INTEGER NOT NULL,

crates/collab/migrations/20221111092550_reconnection_support.sql

@@ -10,8 +10,8 @@ ALTER TABLE "projects"
     DROP COLUMN "unregistered";
 
 CREATE TABLE "worktrees" (
-    "id" INTEGER NOT NULL,
     "project_id" INTEGER NOT NULL REFERENCES projects (id) ON DELETE CASCADE,
+    "id" INTEGER NOT NULL,
     "root_name" VARCHAR NOT NULL,
     "abs_path" VARCHAR NOT NULL,
     "visible" BOOL NOT NULL,
@@ -22,9 +22,9 @@ CREATE TABLE "worktrees" (
 CREATE INDEX "index_worktrees_on_project_id" ON "worktrees" ("project_id");
 
 CREATE TABLE "worktree_entries" (
-    "id" INTEGER NOT NULL,
     "project_id" INTEGER NOT NULL,
     "worktree_id" INTEGER NOT NULL,
+    "id" INTEGER NOT NULL,
     "is_dir" BOOL NOT NULL,
     "path" VARCHAR NOT NULL,
     "inode" INTEGER NOT NULL,
@@ -38,9 +38,9 @@ CREATE TABLE "worktree_entries" (
 CREATE INDEX "index_worktree_entries_on_project_id_and_worktree_id" ON "worktree_entries" ("project_id", "worktree_id");
 
 CREATE TABLE "worktree_diagnostic_summaries" (
-    "path" VARCHAR NOT NULL,
     "project_id" INTEGER NOT NULL,
     "worktree_id" INTEGER NOT NULL,
+    "path" VARCHAR NOT NULL,
     "language_server_id" INTEGER NOT NULL,
     "error_count" INTEGER NOT NULL,
     "warning_count" INTEGER NOT NULL,
@@ -50,8 +50,8 @@ CREATE TABLE "worktree_diagnostic_summaries" (
 CREATE INDEX "index_worktree_diagnostic_summaries_on_project_id_and_worktree_id" ON "worktree_diagnostic_summaries" ("project_id", "worktree_id");
 
 CREATE TABLE "language_servers" (
-    "id" INTEGER NOT NULL,
     "project_id" INTEGER NOT NULL REFERENCES projects (id) ON DELETE CASCADE,
+    "id" INTEGER NOT NULL,
     "name" VARCHAR NOT NULL,
     PRIMARY KEY(project_id, id)
 );

crates/collab/src/db.rs

@@ -1799,6 +1799,68 @@ where
         .await
     }
 
+    pub async fn start_language_server(
+        &self,
+        update: &proto::StartLanguageServer,
+        connection_id: ConnectionId,
+    ) -> Result<Vec<ConnectionId>> {
+        self.transact(|mut tx| async {
+            let project_id = ProjectId::from_proto(update.project_id);
+            let server = update
+                .server
+                .as_ref()
+                .ok_or_else(|| anyhow!("invalid language server"))?;
+
+            // Ensure the update comes from the host.
+            sqlx::query(
+                "
+                SELECT 1
+                FROM projects
+                WHERE id = $1 AND host_connection_id = $2
+                ",
+            )
+            .bind(project_id)
+            .bind(connection_id.0 as i32)
+            .fetch_one(&mut tx)
+            .await?;
+
+            // Add the newly-started language server.
+            sqlx::query(
+                "
+                INSERT INTO language_servers (project_id, id, name)
+                VALUES ($1, $2, $3)
+                ON CONFLICT (project_id, id) DO UPDATE SET
+                    name = excluded.name
+                ",
+            )
+            .bind(project_id)
+            .bind(server.id as i64)
+            .bind(&server.name)
+            .execute(&mut tx)
+            .await?;
+
+            let connection_ids = sqlx::query_scalar::<_, i32>(
+                "
+                SELECT connection_id
+                FROM project_collaborators
+                WHERE project_id = $1 AND connection_id != $2
+                ",
+            )
+            .bind(project_id)
+            .bind(connection_id.0 as i32)
+            .fetch_all(&mut tx)
+            .await?;
+
+            tx.commit().await?;
+
+            Ok(connection_ids
+                .into_iter()
+                .map(|connection_id| ConnectionId(connection_id as u32))
+                .collect())
+        })
+        .await
+    }
+
     pub async fn join_project(
         &self,
         project_id: ProjectId,

crates/collab/src/rpc.rs

@@ -1152,18 +1152,15 @@ impl Server {
         self: Arc<Server>,
         request: Message<proto::StartLanguageServer>,
     ) -> Result<()> {
-        let receiver_ids = self.store().await.start_language_server(
+        let guest_connection_ids = self
-            ProjectId::from_proto(request.payload.project_id),
+            .app_state
-            request.sender_connection_id,
+            .db
-            request
+            .start_language_server(&request.payload, request.sender_connection_id)
-                .payload
+            .await?;
-                .server
+
-                .clone()
-                .ok_or_else(|| anyhow!("invalid language server"))?,
-        )?;
         broadcast(
             request.sender_connection_id,
-            receiver_ids,
+            guest_connection_ids,
             |connection_id| {
                 self.peer.forward_send(
                     request.sender_connection_id,

crates/collab/src/rpc/store.rs

@@ -251,24 +251,6 @@ impl Store {
         }
     }
 
-    pub fn start_language_server(
-        &mut self,
-        project_id: ProjectId,
-        connection_id: ConnectionId,
-        language_server: proto::LanguageServer,
-    ) -> Result<Vec<ConnectionId>> {
-        let project = self
-            .projects
-            .get_mut(&project_id)
-            .ok_or_else(|| anyhow!("no such project"))?;
-        if project.host_connection_id == connection_id {
-            project.language_servers.push(language_server);
-            return Ok(project.connection_ids());
-        }
-
-        Err(anyhow!("no such project"))?
-    }
-
     pub fn leave_project(
         &mut self,
         project_id: ProjectId,