Replace language server WebSocket connection with Ydoc (#14438)

- close #14142

PR replaces binary and JSON-RPC WebSockets with Yjs based communication.

```
+------------------+                +--------------------------------------------------------------+
|                  |                |                                                              |
| +--------------+ |    WebSocket   | +-----------------+   YjsChannel.Server  +-----------------+ |
| |  IDE Client  | |<-------------->| |  Ydoc Server    |<-------------------->| Language Server | |
| +--------------+ |                | +-----------------+                      +-----------------+ |
|     Electron     |                |                         GraalVM                              |
+------------------+                +--------------------------------------------------------------+
```

Workflow:
- IDE client creates a `YjsChannel` backed by Yjs `Y.Array` allowing bidirectional message passing
- IDE client connects to the Ydoc server using the [Yjs WebSocket provider](https://docs.yjs.dev/ecosystem/connection-provider/y-websocket)
- Ydoc server creates a corresponding `YjsChannel` that is synchronized with the client's channel and passes it to the Language server via `YjsChannelCallbacks` callback

This way IDE client and the Language Server establish `YjsChannel` connection and start exchanging messages.

Author: 4e6

.bazelignore

@@ -8,6 +8,7 @@ app/lezer-markdown/node_modules
 app/project-manager-shim/node_modules
 app/table-expression/node_modules
 app/rust-ffi/node_modules
+app/ydoc-channel/node_modules
 app/ydoc-server/node_modules
 app/ydoc-server-nodejs/node_modules
 app/ydoc-server-polyglot/node_modules

BUILD.bazel

@@ -53,6 +53,7 @@ write_source_files(
         "//app/lezer-markdown:write_tsconfigs",
         "//app/project-manager-shim:write_tsconfigs",
         "//app/table-expression:write_tsconfigs",
+        "//app/ydoc-channel:write_tsconfigs",
         "//app/ydoc-server:write_tsconfigs",
         "//app/ydoc-shared:write_tsconfigs",
     ],
@@ -176,6 +177,7 @@ ENGINE_DIST_SOURCES = STDLIB_SOURCES + SBT_PROJECT_FILES + [
         "lib/java/poi-wrapper/src/**",
         "lib/java/runtime-utils/src/**",
         "lib/java/scala-libs-wrapper/src/**",
+        "lib/java/ydoc-api/src/**",
         "lib/java/ydoc-polyfill/src/**",
         "lib/java/ydoc-server/src/**",
         "lib/java/ydoc-server-registration/src/**",

app/gui/integration-test/mock/dataServer.ts

@@ -106,7 +106,16 @@ export function mockDataWSHandler(
           response = createError(builder, LanguageServerErrorCode.NotFile, 'Invalid Path')
           break
         }
-        const file = await readFile(pathSegments(path))
+        let segments: (string | null)[]
+        try {
+          segments = pathSegments(path)
+        } catch {
+          // If path parsing fails, return file not found
+          // This can happen with malformed messages (e.g., visualization data being misinterpreted)
+          response = createError(builder, LanguageServerErrorCode.FileNotFound, 'File not found')
+          break
+        }
+        const file = await readFile(segments)
         if (!file) {
           response = createError(builder, LanguageServerErrorCode.FileNotFound, 'File not found')
           break

app/gui/integration-test/mock/localApi.ts

@@ -26,6 +26,7 @@ import { test } from 'integration-test/base'
 import { uuidv4 } from 'lib0/random.js'
 import { join } from 'node:path'
 import type { Page, WebSocketRoute } from 'playwright'
+import { YjsChannel } from 'ydoc-channel'
 import { WSSharedDoc, YjsConnection, type YjsSocket } from 'ydoc-server'
 import { makeVisUpdates, mockDataHandler, mockLSHandler, mockYdocProvider } from './lsHandler'
 
@@ -37,6 +38,8 @@ const ROOT_PARENT_PATH = Path('/home/user/enso')
 const ROOT_PATH = Path('/home/user/enso/enso-projects')
 const DOWNLOAD_PATH = Path('/home/user/enso/Downloads')
 
+// These addresses are kept for backward compatibility in OpenProject response
+// but the actual communication now flows through YjsChannels
 const languageServerJsonAddress = { host: '127.0.0.1', port: 1235 }
 const languageServerBinaryAddress = { host: '127.0.0.1', port: 1234 }
 const languageServerYdocAddress = { host: '127.0.0.1', port: 1233 }
@@ -241,7 +244,8 @@ export async function mockLocalApi(page: Page) {
   addDirectory({ path: ROOT_PATH })
   addDirectory({ path: DOWNLOAD_PATH })
 
-  let languageServerBinaryWs: WebSocketRoute | null = null
+  // Data channel for visualization updates, set up when client connects
+  let dataChannel: YjsChannel<Uint8Array> | null = null
 
   await test.step('Mock Local API', async () => {
     const toJSONRPCResult = (result: unknown): JSONRPCResponse<unknown> => ({
@@ -340,40 +344,6 @@ export async function mockLocalApi(page: Page) {
       })
     })
 
-    await page.routeWebSocket(
-      `ws://${languageServerBinaryAddress.host}:${languageServerBinaryAddress.port}/`,
-      (ws) => {
-        languageServerBinaryWs = ws
-        ws.onMessage(async (messageRaw) => {
-          const response = await mockDataHandler(new Uint8Array(Buffer.from(messageRaw)).buffer)
-          if (response) {
-            ws.send(Buffer.from(response))
-          }
-        })
-      },
-    )
-    await page.routeWebSocket(
-      `ws://${languageServerJsonAddress.host}:${languageServerJsonAddress.port}/`,
-      (ws) => {
-        ws.onMessage(async (messageRaw) => {
-          const { method, params, jsonrpc, id } = JSON.parse(messageRaw.toString())
-          try {
-            const result =
-              (await mockLSHandler(
-                method,
-                params,
-                (message) => ws.send(JSON.stringify({ jsonrpc, ...message })),
-                (binaryData?: ArrayBuffer) => {
-                  if (binaryData) languageServerBinaryWs?.send(Buffer.from(binaryData))
-                },
-              )) ?? null
-            ws.send(JSON.stringify({ jsonrpc, id, result }))
-          } catch (error) {
-            ws.send(JSON.stringify({ jsonrpc, id, error }))
-          }
-        })
-      },
-    )
     const ydocAddressBase = `ws://${languageServerYdocAddress.host}:${languageServerYdocAddress.port}`
 
     class MockWs implements YjsSocket {
@@ -413,14 +383,76 @@ export async function mockLocalApi(page: Page) {
       }
     }
 
+    /** Set up the mock JSON-RPC channel handler with its associated data channel */
+    function setupMockLsChannel(
+      channel: YjsChannel<string>,
+      associatedDataChannel: YjsChannel<Uint8Array> | null,
+    ) {
+      channel.subscribe(async (messageRaw) => {
+        const { method, params, jsonrpc, id } = JSON.parse(messageRaw)
+        try {
+          const result =
+            (await mockLSHandler(
+              method,
+              params,
+              (message) => channel.send(JSON.stringify({ jsonrpc, ...message })),
+              (binaryData?: ArrayBuffer) => {
+                // Use the associated data channel for this connection
+                if (binaryData && associatedDataChannel) {
+                  associatedDataChannel.send(new Uint8Array(binaryData))
+                }
+              },
+            )) ?? null
+          channel.send(JSON.stringify({ jsonrpc, id, result }))
+        } catch (error) {
+          channel.send(JSON.stringify({ jsonrpc, id, error }))
+        }
+      })
+    }
+
+    /** Set up the mock binary data channel handler */
+    function setupMockDataChannel(channel: YjsChannel<Uint8Array>) {
+      // Also set the global dataChannel for updateVisualization API
+      dataChannel = channel
+      channel.subscribe(async (messageRaw) => {
+        // Important: Use slice to get a copy of just the relevant portion
+        // because messageRaw.buffer may include data beyond the Uint8Array's view
+        const data = messageRaw.buffer.slice(
+          messageRaw.byteOffset,
+          messageRaw.byteOffset + messageRaw.byteLength,
+        ) as ArrayBuffer
+        const response = await mockDataHandler(data)
+        if (response) {
+          channel.send(new Uint8Array(response))
+        }
+      })
+    }
+
     await page.routeWebSocket(`${ydocAddressBase}/**`, (wsRoute) => {
       const parsedUrl = new URL(wsRoute.url())
       const room = parsedUrl.pathname.substring('/project/'.length)
+      const lsChannelName = parsedUrl.searchParams.get('ls')
+      const dataChannelName = parsedUrl.searchParams.get('data')
 
       const mockWs = new MockWs(wsRoute)
       const wsDoc = new WSSharedDoc()
       const _connection = new YjsConnection(mockWs, wsDoc)
       mockYdocProvider(room, wsDoc.doc)
+
+      let binaryChannel: YjsChannel<Uint8Array> | null = null
+      if (dataChannelName) {
+        binaryChannel = new YjsChannel<Uint8Array>(wsDoc.doc, dataChannelName)
+        // Only set the global dataChannel for the main 'index' room connection.
+        // Subdoc connections should not overwrite it, as the client's DataServer
+        // only listens on the main document's channel.
+        if (room === 'index') {
+          setupMockDataChannel(binaryChannel)
+        }
+      }
+      if (lsChannelName) {
+        const lsChannel = new YjsChannel<string>(wsDoc.doc, lsChannelName)
+        setupMockLsChannel(lsChannel, binaryChannel)
+      }
     })
 
     await page.route('/api/root-directory-path', async (route, request) => {
@@ -638,7 +670,7 @@ export async function mockLocalApi(page: Page) {
 
   async function updateVisualization(preprocessor: string, data: unknown) {
     for (const update of makeVisUpdates(preprocessor, data)) {
-      languageServerBinaryWs?.send(Buffer.from(update))
+      dataChannel?.send(new Uint8Array(update))
     }
   }
 

app/gui/package.json

@@ -126,6 +126,7 @@
     "y-protocols": "^1.0.6",
     "y-textarea": "^1.0.2",
     "y-websocket": "^1.5.4",
+    "ydoc-channel": "workspace:*",
     "ydoc-shared": "workspace:*",
     "yjs": "^13.6.21",
     "zod": "catalog:",

app/gui/src/project-view/util/crdt.ts

@@ -47,6 +47,8 @@ interface SubdocsEvent {
 export type ProviderParams = {
   /** URL for the project's language server RPC connection. */
   ls: string
+  /** URL for the project's data connection. */
+  data: string
 }
 
 /** TODO: Add docs */

app/gui/src/project-view/util/net.ts

@@ -1,27 +1,19 @@
 import { onScopeDispose } from 'vue'
+import { YjsChannel } from 'ydoc-channel'
 import { AbortScope } from 'ydoc-shared/util/net'
-import {
-  ReconnectingWebSocket,
-  ReconnectingWebSocketTransport,
-} from 'ydoc-shared/util/net/ReconnectingWSTransport'
+import { YjsTransport } from 'ydoc-shared/util/net/YjsTransport'
+import * as Y from 'yjs'
 
 export { AbortScope }
 
-const WS_OPTIONS = {
-  // We do not want to enqueue any messages, because after reconnecting we have to initProtocol again.
-  maxEnqueuedMessages: 0,
-}
-
 /** TODO: Add docs */
-export function createRpcTransport(url: string): ReconnectingWebSocketTransport {
-  return new ReconnectingWebSocketTransport(url, WS_OPTIONS)
+export function createRpcTransport(indexDoc: Y.Doc, url: string): YjsTransport {
+  return new YjsTransport(indexDoc, url)
 }
 
 /** TODO: Add docs */
-export function createDataWebsocket(url: string, binaryType: 'arraybuffer' | 'blob'): WebSocket {
-  const websocket = new ReconnectingWebSocket(url, undefined, WS_OPTIONS)
-  websocket.binaryType = binaryType
-  return websocket as WebSocket
+export function createDataSocket(indexDoc: Y.Doc, url: string): YjsChannel {
+  return new YjsChannel(indexDoc, url)
 }
 
 export interface WebSocketHandler {

app/gui/src/project-view/util/net/dataServer.ts

@@ -1,5 +1,6 @@
 import { Err, Ok, type Result } from 'enso-common/src/utilities/data/result'
 import { ObservableV2 } from 'lib0/observable'
+import type { YjsChannel } from 'ydoc-channel'
 import {
   Builder,
   ByteBuffer,
@@ -57,23 +58,30 @@ export class DataServer extends ObservableV2<DataServerEvents> {
   private initializationScheduled = false
   resolveCallbacks = new Map<string, (data: any) => void>()
 
-  /** `websocket.binaryType` should be `ArrayBuffer`. */
+  /** TODO: Add docs */
   constructor(
     public clientId: string,
-    public websocket: WebSocket,
+    public channel: YjsChannel,
     abort: AbortScope,
   ) {
     super()
     abort.handleDispose(this)
 
-    websocket.addEventListener('message', ({ data: rawPayload }) => {
-      if (!(rawPayload instanceof ArrayBuffer)) {
+    channel.addEventListener('message', ({ data: rawPayload }) => {
+      if (!ArrayBuffer.isView(rawPayload)) {
         console.warn('Data Server: Data type was invalid:', rawPayload)
         // Ignore all non-binary messages. If the messages are `Blob`s instead, this is a
         // misconfiguration and should also be ignored.
         return
       }
-      const binaryMessage = OutboundMessage.getRootAsOutboundMessage(new ByteBuffer(rawPayload))
+      const binaryMessage = OutboundMessage.getRootAsOutboundMessage(
+        new ByteBuffer(
+          rawPayload.buffer.slice(
+            rawPayload.byteOffset,
+            rawPayload.byteOffset + rawPayload.byteLength,
+          ),
+        ),
+      )
       const payloadType = binaryMessage.payloadType()
       const payload = binaryMessage.payload(new PAYLOAD_CONSTRUCTOR[payloadType]())
       if (!payload) return
@@ -94,20 +102,19 @@ export class DataServer extends ObservableV2<DataServerEvents> {
         callback?.(payload)
       }
     })
-    websocket.addEventListener('error', (error) =>
+    channel.addEventListener('error', (error) =>
       console.error('Language Server Binary socket error:', error),
     )
-    websocket.addEventListener('close', () => {
+    channel.addEventListener('close', () => {
       this.scheduleInitializationAfterConnect()
     })
 
-    if (websocket.readyState === WebSocket.OPEN) this.initialized = this.initialize()
-    else this.initialized = this.scheduleInitializationAfterConnect()
+    this.initialized = this.initialize()
   }
 
   /** TODO: Add docs */
   dispose() {
-    this.websocket.close()
+    this.channel.close()
     this.resolveCallbacks.clear()
   }
 
@@ -116,11 +123,11 @@ export class DataServer extends ObservableV2<DataServerEvents> {
     this.initializationScheduled = true
     this.initialized = new Promise((resolve) => {
       const cb = () => {
-        this.websocket.removeEventListener('open', cb)
+        this.channel.removeEventListener('open', cb)
         this.initializationScheduled = false
         resolve(this.initialize())
       }
-      this.websocket.addEventListener('open', cb)
+      this.channel.addEventListener('open', cb)
     })
     return this.initialized
   }
@@ -166,7 +173,7 @@ export class DataServer extends ObservableV2<DataServerEvents> {
       this.resolveCallbacks.set(messageUuid, resolve)
     })
     try {
-      this.websocket.send(builder.finish(rootTable).toArrayBuffer())
+      this.channel.send(builder.finish(rootTable).toArrayBuffer())
     } catch (e: unknown) {
       this.resolveCallbacks.delete(messageUuid)
       throw e