fix(backend/lsp): Send text operations to rust-analyzer

Author: Brayan-724

backend/runner/src/lsp.rs

@@ -19,7 +19,7 @@ pub struct LspInput;
 
 impl LspInput {
     pub fn notify_value<T: Notification>(
-        params: T::Params,
+        params: &T::Params,
     ) -> Result<serde_json::Value, serde_json::Error> {
         Ok(serde_json::json!({
             "jsonrpc": JsonRpcVersion,
@@ -28,7 +28,7 @@ impl LspInput {
         }))
     }
 
-    pub fn notify<T: Notification>(params: T::Params) -> Result<String, serde_json::Error> {
+    pub fn notify<T: Notification>(params: &T::Params) -> Result<String, serde_json::Error> {
         serde_json::to_string(&Self::notify_value::<T>(params)?)
     }
 

backend/src/project/project.rs

@@ -1,15 +1,22 @@
 use std::collections::{HashMap, HashSet};
+use std::str::FromStr;
 use std::sync::Arc;
 
 use actix::Addr;
 use futures::StreamExt;
+use rsground_runner::lsp::notification::{DidChangeTextDocument, DidOpenTextDocument};
+use rsground_runner::lsp::{
+    DidChangeTextDocumentParams, DidOpenTextDocumentParams, TextDocumentContentChangeEvent,
+    TextDocumentItem, Uri, VersionedTextDocumentIdentifier,
+};
 use rsground_runner::Runner;
 use tokio::sync::broadcast;
 use uuid::Uuid;
 
 use crate::auth::jwt::RgUserData;
 use crate::collab::{Document, DocumentInfo};
 use crate::http_errors::HttpErrors;
+use crate::project::project_lsp::ProjectLspActor;
 use crate::utils::{ArcStr, AsyncDefault, AsyncInto, ToStream, EMPTY_STR};
 use crate::ws::messages::{InternalMessage, ServerMessage};
 
@@ -29,6 +36,7 @@ pub struct Project {
     runner: Arc<Runner>,
     producer: Addr<producer::ProjectProducer>,
     lsp: Addr<lsp::ProjectLsp>,
+    lsp_started: bool,
 }
 
 impl AsyncDefault for Project {
@@ -58,6 +66,7 @@ impl AsyncDefault for Project {
             runner,
             producer,
             lsp,
+            lsp_started: false,
         }
     }
 }
@@ -79,8 +88,33 @@ impl Project {
         &self.lsp
     }
 
-    pub async fn start_lsp(&self) {
+    pub async fn start_lsp(&mut self) {
+        if self.lsp_started {
+            return;
+        }
+
+        self.lsp_started = true;
+
         self.lsp.do_send(lsp::Execute);
+
+        for (file, doc) in &self.documents {
+            let version = doc.revision().await as i32;
+            let text = doc.text().await;
+
+            let Ok(uri) = Uri::from_str(&format!("file:///home/{file}")) else {
+                continue;
+            };
+
+            self.lsp
+                .send_notify::<DidOpenTextDocument, true>(&DidOpenTextDocumentParams {
+                    text_document: TextDocumentItem {
+                        uri,
+                        language_id: "rust".to_owned(),
+                        version,
+                        text,
+                    },
+                });
+        }
     }
 
     pub async fn execute(&self) {
@@ -110,13 +144,26 @@ impl Project {
 
     pub async fn add_file(&mut self, path: impl Into<ArcStr>, document: Document) -> Arc<Document> {
         let path: ArcStr = path.into();
+        let content = document.text().await;
 
         _ = self
             .get_runner()
-            .create_file(&path.to_string(), &document.text().await)
+            .create_file(&path.to_string(), &content)
             .await
             .inspect_err(|err| project_log::error!("{err}"));
 
+        if let Ok(uri) = Uri::from_str(&format!("file:///home/{path}")) {
+            self.lsp
+                .send_notify::<DidOpenTextDocument, true>(&DidOpenTextDocumentParams {
+                    text_document: TextDocumentItem {
+                        uri,
+                        language_id: "rust".to_owned(),
+                        version: 0,
+                        text: content,
+                    },
+                });
+        }
+
         self.documents
             .entry(path)
             .insert_entry(document.into())
@@ -125,9 +172,38 @@ impl Project {
     }
 
     pub fn rm_file(&mut self, path: impl AsRef<str>) -> Option<Arc<Document>> {
+        // TODO: (@Brayan-724) Remove real file and notify lsp
         self.documents.remove(path.as_ref())
     }
 
+    pub async fn file_edit(&mut self, path: ArcStr, content: impl AsRef<str>) {
+        _ = self
+            .runner
+            .create_file(&path, content.as_ref())
+            .await
+            .inspect_err(|err| log::error!("{err}"));
+
+        if let Ok(uri) = Uri::from_str(&format!("file:///home/{path}")) {
+            let version = if let Some(doc) = self.documents.get(&path) {
+                doc.revision().await as i32
+            } else {
+                0
+            };
+
+            self.lsp
+                .send_notify::<DidChangeTextDocument, true>(&DidChangeTextDocumentParams {
+                    text_document: VersionedTextDocumentIdentifier { uri, version },
+                    content_changes: vec![TextDocumentContentChangeEvent {
+                        range: None,
+                        range_length: None,
+                        text: content.as_ref().to_owned(),
+                    }],
+                });
+        }
+
+        _ = self.internal.send(InternalMessage::FileEdit { path });
+    }
+
     /// Get all file paths
     pub async fn get_files(&self) -> HashMap<ArcStr, DocumentInfo> {
         (&self.documents)

backend/src/project/project_lsp.rs

@@ -1,6 +1,7 @@
 mod client;
 mod server;
 
+use std::collections::VecDeque;
 use std::io::{self, Write};
 use std::sync::Arc;
 
@@ -17,7 +18,7 @@ use rsground_runner::{PipeWriter, Runner};
 use tokio::sync::broadcast;
 use uuid::Uuid;
 
-use crate::utils::{define_local_logger, ArcStr};
+use crate::utils::{define_local_logger, ArcStr, Truncate};
 use crate::ws::messages::InternalMessage;
 
 use super::project_runner::{start_job, AbortSender};
@@ -36,10 +37,12 @@ define_local_logger!(local_log as "backend::lsp" {
 const LSP_INITIALIZATION_ID: &str = "rsground::initialize";
 
 pub struct ProjectLsp {
-    project_id: Uuid,
     broadcast: broadcast::Sender<InternalMessage>,
-    runner: Arc<Runner>,
     instance: Option<(AbortSender, PipeWriter)>,
+    project_id: Uuid,
+    queue: VecDeque<Stdin>,
+    ready: bool,
+    runner: Arc<Runner>,
 }
 
 impl ProjectLsp {
@@ -49,10 +52,12 @@ impl ProjectLsp {
         runner: Arc<Runner>,
     ) -> Addr<Self> {
         let project_lsp = Self {
-            project_id,
             broadcast,
-            runner,
             instance: None,
+            project_id,
+            queue: VecDeque::new(),
+            ready: false,
+            runner,
         };
 
         project_lsp.start()
@@ -63,6 +68,39 @@ impl Actor for ProjectLsp {
     type Context = Context<ProjectLsp>;
 }
 
+pub trait ProjectLspActor {
+    fn send_request<T: Request, const QUEUE: bool>(
+        &self,
+        id: impl serde::Serialize,
+        params: &T::Params,
+    );
+    fn send_notify<T: Notification, const QUEUE: bool>(&self, params: &T::Params);
+}
+
+impl ProjectLspActor for Addr<ProjectLsp> {
+    fn send_request<T: Request, const QUEUE: bool>(
+        &self,
+        id: impl serde::Serialize,
+        params: &T::Params,
+    ) {
+        match LspInput::request::<T>(id, params) {
+            Ok(req) => self.do_send(Stdin::<QUEUE>(req)),
+            Err(err) => {
+                local_log::stdin::error!("Serialize: {err:?}");
+            }
+        }
+    }
+
+    fn send_notify<T: Notification, const QUEUE: bool>(&self, params: &T::Params) {
+        match LspInput::notify::<T>(params) {
+            Ok(req) => self.do_send(Stdin::<QUEUE>(req)),
+            Err(err) => {
+                local_log::stdin::error!("Serialize: {err:?}");
+            }
+        }
+    }
+}
+
 // Client-side messages
 
 #[derive(Message)]
@@ -71,7 +109,7 @@ pub struct Execute;
 
 #[derive(Message)]
 #[rtype(result = "()")]
-pub struct Stdin(String);
+pub struct Stdin<const QUEUE: bool = true>(String);
 
 #[derive(Message)]
 #[rtype(result = "()")]
@@ -103,19 +141,10 @@ impl Handler<Execute> for ProjectLsp {
             move |abort, this, ctx| {
                 this.instance.replace((abort, stdin));
 
-                match LspInput::request::<Initialize>(
+                ctx.address().send_request::<Initialize, false>(
                     LSP_INITIALIZATION_ID,
                     &*client::LSP_INITIALIZATION,
-                ) {
-                    Ok(req) => {
-                        local_log::initialize::info!(target: this.project_id, "Initialize sended");
-                        ctx.notify(Stdin(req))
-                    }
-                    Err(err) => {
-                        local_log::initialize::error!(target: this.project_id, "{err:?}");
-                        ctx.notify(Kill);
-                    }
-                }
+                );
             },
             async move |abort| child.wait_or_abort(abort).await,
             |_, this, _| {
@@ -132,21 +161,26 @@ impl Handler<Kill> for ProjectLsp {
     type Result = ();
 
     fn handle(&mut self, _: Kill, _: &mut Self::Context) -> Self::Result {
+        self.ready = false;
         if let Some((abort, _)) = self.instance.take() {
             _ = abort.send(());
         }
     }
 }
 
-impl Handler<Stdin> for ProjectLsp {
+impl<const QUEUE: bool> Handler<Stdin<QUEUE>> for ProjectLsp {
     type Result = ();
 
-    fn handle(&mut self, Stdin(msg): Stdin, _: &mut Self::Context) -> Self::Result {
-        if let Some((_, stdin)) = self.instance.as_mut() {
-            let msg = format!("Content-Length: {}\r\n\r\n{msg}", msg.len());
-            if let Err(err) = stdin.write_all(msg.as_bytes()).and_then(|_| stdin.flush()) {
-                local_log::stdin::error!(target: self.project_id, "Cannot write: {err}")
+    fn handle(&mut self, Stdin(msg): Stdin<QUEUE>, _: &mut Self::Context) -> Self::Result {
+        if !QUEUE || self.ready {
+            if let Some((_, stdin)) = self.instance.as_mut() {
+                let msg = format!("Content-Length: {}\r\n\r\n{msg}", msg.len());
+                if let Err(err) = stdin.write_all(msg.as_bytes()).and_then(|_| stdin.flush()) {
+                    local_log::stdin::error!(target: self.project_id, "Cannot write: {err}")
+                }
             }
+        } else {
+            self.queue.push_back(Stdin(msg));
         }
     }
 }
@@ -229,7 +263,7 @@ impl Handler<ClientStdin> for ProjectLsp {
             return;
         };
 
-        ctx.notify(Stdin(msg));
+        ctx.notify(Stdin::<true>(msg));
     }
 }
 
@@ -248,9 +282,16 @@ impl StreamHandler<Stdout> for ProjectLsp {
             LspOutput::Response(res) if res.id() == LSP_INITIALIZATION_ID => match res {
                 LspResponse::Ok { result, .. } => {
                     local_log::initialize::info!(target: self.project_id, "Initialize received");
-                    local_log::initialize::trace!(target: self.project_id, "{result:?}");
-                    if let Ok(noti) = LspInput::notify::<Initialized>(InitializedParams {}) {
-                        ctx.notify(Stdin(noti));
+                    local_log::initialize::trace!(target: self.project_id, "{:?}", result.truncate::<256>());
+
+                    self.ready = true;
+
+                    if let Ok(noti) = LspInput::notify::<Initialized>(&InitializedParams {}) {
+                        ctx.notify(Stdin::<false>(noti));
+                    }
+
+                    while let Some(msg) = self.queue.pop_front() {
+                        ctx.notify(msg);
                     }
                 }
                 LspResponse::Err { error, .. } => {
@@ -260,7 +301,7 @@ impl StreamHandler<Stdout> for ProjectLsp {
             },
 
             LspOutput::Response(res) => {
-                local_log::response::trace!(target: self.project_id, "{res:?}");
+                local_log::response::trace!(target: self.project_id, "{:?}", (&res).truncate::<256>());
                 let (client_id, res_id) = res
                     .id()
                     .clone()

backend/src/project/project_lsp/server.rs

@@ -111,7 +111,7 @@ pub const LSP_INITIALIZATION: LazyLock<InitializeResult> = LazyLock::new(|| Init
         }),
         text_document_sync: Some(TextDocumentSyncCapability::Options(
             TextDocumentSyncOptions {
-                change: Some(TextDocumentSyncKind::INCREMENTAL),
+                change: Some(TextDocumentSyncKind::FULL),
                 open_close: Some(true),
                 save: Some(TextDocumentSyncSaveOptions::SaveOptions(
                     SaveOptions::default(),

backend/src/ws/handlers.rs

@@ -367,7 +367,6 @@ impl RgWebsocket {
 
                 let project = self.app_state.get_project(self.project_id).await?;
                 let mut project = project.write().await;
-                let runner = project.get_runner();
 
                 let doc = project.get_file(&file).ok_or_else(|| {
                     local_log::sync::error!("File {file:?} not found in {:?}", self.project_id);
@@ -383,14 +382,7 @@ impl RgWebsocket {
                         .send(ServerMessage::Error { message: err })
                 }
 
-                _ = runner
-                    .create_file(&file, &doc.text().await)
-                    .await
-                    .inspect_err(|err| log::error!("{err}"));
-
-                _ = project
-                    .internal
-                    .send(InternalMessage::FileEdit { path: file });
+                project.file_edit(file, doc.text().await).await;
 
                 Err(ServerMessageError::None)
             }