Replace language server WebSocket connection with Ydoc

.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

@@ -141,6 +141,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/package.json

@@ -124,6 +124,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,
@@ -60,20 +61,22 @@ export class DataServer extends ObservableV2<DataServerEvents> {
   /** `websocket.binaryType` should be `ArrayBuffer`. */
   constructor(
     public clientId: string,
-    public websocket: WebSocket,
+    public websocket: YjsChannel,
     abort: AbortScope,
   ) {
     super()
     abort.handleDispose(this)
 
     websocket.addEventListener('message', ({ data: rawPayload }) => {
-      if (!(rawPayload instanceof ArrayBuffer)) {
+      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),
+      )
       const payloadType = binaryMessage.payloadType()
       const payload = binaryMessage.payload(new PAYLOAD_CONSTRUCTOR[payloadType]())
       if (!payload) return
@@ -101,8 +104,7 @@ export class DataServer extends ObservableV2<DataServerEvents> {
       this.scheduleInitializationAfterConnect()
     })
 
-    if (websocket.readyState === WebSocket.OPEN) this.initialized = this.initialize()
-    else this.initialized = this.scheduleInitializationAfterConnect()
+    this.initialized = this.initialize()
   }
 
   /** TODO: Add docs */

app/gui/src/providers/openedProjects/project/project.ts

@@ -14,7 +14,7 @@ import { nextEvent } from '@/util/data/observable'
 import type { Opt } from '@/util/data/opt'
 import { ReactiveMapping } from '@/util/database/reactiveDb'
 import type { MethodPointer } from '@/util/methodPointer'
-import { createDataWebsocket, createRpcTransport, useAbortScope } from '@/util/net'
+import { createDataSocket, createRpcTransport, useAbortScope } from '@/util/net'
 import { DataServer } from '@/util/net/dataServer'
 import { ProjectPath } from '@/util/projectPath'
 import { tryQualifiedName, type QualifiedName } from '@/util/qualifiedName'
@@ -74,15 +74,17 @@ export function createProjectStore(
 
   const doc = new Y.Doc()
   const awareness = new Awareness(doc)
-
+  const ydocUrl = resolveYDocUrl(props.engine.rpcUrl, props.engine.ydocUrl)
+  const guiRpcId = `gui-rpc-${crypto.randomUUID()}`
   const clientId = crypto.randomUUID() as Uuid
-  const lsRpcConnection = createLsRpcConnection(clientId, props.engine.rpcUrl, abort)
+  const lsRpcConnection = createLsRpcConnection(clientId, doc, guiRpcId, abort)
   const projectRootId = lsRpcConnection.contentRoots.then(
     (roots) => roots.find((root) => root.type === 'Project')?.id,
   )
   onScopeDispose(() => lsRpcConnection.release())
 
-  const dataConnection = initializeDataConnection(clientId, props.engine.dataUrl, abort)
+  const guiDataId = `gui-data-${crypto.randomUUID()}`
+  const dataConnection = initializeDataConnection(clientId, doc, guiDataId, abort)
   const rpcUrl = new URL(props.engine.rpcUrl)
   const isOnLocalBackend =
     rpcUrl.protocol === 'mock:' ||
@@ -91,23 +93,12 @@ export function createProjectStore(
     rpcUrl.hostname === '[::1]' ||
     rpcUrl.hostname === '0:0:0:0:0:0:0:1'
 
-  const moduleProjectPath = computed((): Result<ProjectPath> | undefined => {
-    const filePath = observedFileName.value
-    if (filePath == null) return undefined
-    const withoutFileExt = filePath.replace(/\.enso$/, '')
-    const withDotSeparators = withoutFileExt.replace(/\//g, '.')
-    const qn = tryQualifiedName(withDotSeparators)
-    if (!qn.ok) return qn
-    return Ok(ProjectPath.create(undefined, qn.value))
-  })
-
-  const ydocUrl = resolveYDocUrl(props.engine.rpcUrl, props.engine.ydocUrl)
   let yDocsProvider: ReturnType<typeof attachProvider> | undefined
   watchEffect((onCleanup) => {
     yDocsProvider = attachProvider(
       ydocUrl.href,
       'index',
-      { ls: props.engine.rpcUrl },
+      { ls: guiRpcId, data: guiDataId },
       doc,
       awareness.internal,
     )
@@ -117,6 +108,16 @@ export function createProjectStore(
     })
   })
 
+  const moduleProjectPath = computed((): Result<ProjectPath> | undefined => {
+    const filePath = observedFileName.value
+    if (filePath == null) return undefined
+    const withoutFileExt = filePath.replace(/\.enso$/, '')
+    const withDotSeparators = withoutFileExt.replace(/\//g, '.')
+    const qn = tryQualifiedName(withDotSeparators)
+    if (!qn.ok) return qn
+    return Ok(ProjectPath.create(undefined, qn.value))
+  })
+
   const projectModel = new DistributedProject(doc)
 
   const entryPoint = computed<MethodPointer>(() => {
@@ -443,8 +444,13 @@ function resolveYDocUrl(rpcUrl: string, url: string): URL {
   return resolved
 }
 
-function createLsRpcConnection(clientId: Uuid, url: string, abort: AbortScope): LanguageServer {
-  const transport = createRpcTransport(url)
+function createLsRpcConnection(
+  clientId: Uuid,
+  doc: Y.Doc,
+  url: string,
+  abort: AbortScope,
+): LanguageServer {
+  const transport = createRpcTransport(doc, url)
   const connection = new LanguageServer(clientId, transport)
   abort.onAbort(() => {
     connection.stopReconnecting()
@@ -453,8 +459,8 @@ function createLsRpcConnection(clientId: Uuid, url: string, abort: AbortScope):
   return connection
 }
 
-function initializeDataConnection(clientId: Uuid, url: string, abort: AbortScope) {
-  const client = createDataWebsocket(url, 'arraybuffer')
+function initializeDataConnection(clientId: Uuid, doc: Y.Doc, url: string, abort: AbortScope) {
+  const client = createDataSocket(doc, url)
   const connection = new DataServer(clientId, client, abort)
   onScopeDispose(() => connection.dispose())
   return connection

app/project-manager-shim/src/projectService/ensoRunner.ts

@@ -87,8 +87,14 @@ export class EnsoRunner implements Runner {
     args: readonly string[],
     spawnCallback: (cmd: string, cmdArgs: readonly string[]) => T,
   ) {
-    const cmd = this.ensoPath.endsWith('.bat') ? 'cmd.exe' : this.ensoPath
-    const cmdArgs = this.ensoPath.endsWith('.bat') ? ['/c', this.ensoPath, ...args] : args
+    const [cmd, cmdArgs] =
+      this.ensoPath.endsWith('.bat') ?
+        ['cmd.exe', ['/c', this.ensoPath, ...args]]
+      : [this.ensoPath, args]
+    const isDevMode = process.env.NODE_ENV === 'development'
+    if (isDevMode) {
+      console.log('runProcess', cmd, 'with', cmdArgs)
+    }
     return spawnCallback(cmd, cmdArgs)
   }
 

app/ydoc-channel/BUILD.bazel

@@ -0,0 +1,34 @@
+load("@aspect_rules_js//npm:defs.bzl", "npm_package")
+load("@aspect_rules_ts//ts:defs.bzl", "ts_config", "ts_project")
+load("@npm//:defs.bzl", "npm_link_all_packages", "npm_link_targets")
+
+npm_link_all_packages(name = "node_modules")
+
+ts_config(
+    name = "tsconfig",
+    src = "tsconfig.json",
+    deps = ["//:tsconfig"],
+)
+
+ts_project(
+    name = "tsc",
+    srcs = glob(["src/**/*.ts"]),
+    composite = True,
+    out_dir = "dist",
+    root_dir = "src",
+    tsconfig = ":tsconfig",
+    validate = select({
+        "@platforms//os:windows": False,
+        "//conditions:default": True,
+    }),
+    deps = npm_link_targets(),
+)
+
+npm_package(
+    name = "pkg",
+    srcs = [
+        "package.json",
+        ":tsc",
+    ],
+    visibility = ["//visibility:public"],
+)

app/ydoc-channel/package.json

@@ -0,0 +1,28 @@
+{
+  "name": "ydoc-channel",
+  "version": "0.1.0",
+  "description": "Y.js-based bidirectional communication channel",
+  "type": "module",
+  "exports": {
+    ".": {
+      "source": "./src/index.ts",
+      "types": "./dist/index.d.ts",
+      "import": "./dist/index.js"
+    }
+  },
+  "main": "src/index.ts",
+  "scripts": {
+    "test:unit": "vitest run",
+    "compile": "tsc",
+    "lint": "eslint . --cache --max-warnings=0"
+  },
+  "devDependencies": {
+    "@types/node": "catalog:",
+    "typescript": "catalog:",
+    "vitest": "catalog:"
+  },
+  "dependencies": {
+    "lib0": "^0.2.99",
+    "yjs": "^13.6.19"
+  }
+}