refactor: reimplement realtime-communication

This commit refactors a lot of things that are not easy to separate.
It replaces the binary protocol of y-protocols with json.
It introduces event based message processing.
It implements our own code mirror plugins for synchronisation of content and remote cursors

Signed-off-by: Tilman Vatteroth <git@tilmanvatteroth.de>

commons/src/y-doc-sync/in-memory-connection-message.transporter.ts

@@ -0,0 +1,52 @@
+/*
+ * SPDX-FileCopyrightText: 2023 The HedgeDoc developers (see AUTHORS file)
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+import {
+  ConnectionState,
+  MessageTransporter
+} from '../message-transporters/message-transporter.js'
+import { Message, MessageType } from '../message-transporters/message.js'
+
+/**
+ * Message transporter for testing purposes that redirects message to another in memory connection message transporter instance.
+ */
+export class InMemoryConnectionMessageTransporter extends MessageTransporter {
+  private otherSide: InMemoryConnectionMessageTransporter | undefined
+
+  constructor(private name: string) {
+    super()
+  }
+
+  public connect(other: InMemoryConnectionMessageTransporter): void {
+    this.otherSide = other
+    other.otherSide = this
+    this.onConnected()
+    other.onConnected()
+  }
+
+  public disconnect(): void {
+    this.onDisconnecting()
+
+    if (this.otherSide) {
+      this.otherSide.onDisconnecting()
+      this.otherSide.otherSide = undefined
+      this.otherSide = undefined
+    }
+  }
+
+  sendMessage(content: Message<MessageType>): void {
+    if (this.otherSide === undefined) {
+      throw new Error('Disconnected')
+    }
+    console.debug(`${this.name}`, 'Sending', content)
+    this.otherSide?.receiveMessage(content)
+  }
+
+  getConnectionState(): ConnectionState {
+    return this.otherSide !== undefined
+      ? ConnectionState.CONNECTED
+      : ConnectionState.DISCONNECT
+  }
+}

commons/src/y-doc-sync/realtime-doc.test.ts

@@ -0,0 +1,16 @@
+/*
+ * SPDX-FileCopyrightText: 2023 The HedgeDoc developers (see AUTHORS file)
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+import { RealtimeDoc } from './realtime-doc.js'
+import { describe, expect, it } from '@jest/globals'
+
+describe('websocket-doc', () => {
+  it('saves the initial content', () => {
+    const textContent = 'textContent'
+    const websocketDoc = new RealtimeDoc(textContent)
+
+    expect(websocketDoc.getCurrentContent()).toBe(textContent)
+  })
+})

commons/src/y-doc-sync/realtime-doc.ts

@@ -0,0 +1,48 @@
+/*
+ * SPDX-FileCopyrightText: 2023 The HedgeDoc developers (see AUTHORS file)
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+import { Doc } from 'yjs'
+import { Text as YText } from 'yjs'
+
+const MARKDOWN_CONTENT_CHANNEL_NAME = 'markdownContent'
+
+/**
+ * This is the implementation of {@link Doc YDoc} which includes additional handlers for message sending and receiving.
+ */
+export class RealtimeDoc extends Doc {
+  /**
+   * Creates a new instance.
+   *
+   * The new instance is filled with the given initial content.
+   *
+   * @param initialContent - the initial content of the {@link Doc YDoc}
+   */
+  constructor(initialContent?: string) {
+    super()
+    if (initialContent) {
+      this.getMarkdownContentChannel().insert(0, initialContent)
+    }
+  }
+
+  /**
+   * Extracts the {@link YText text channel} that contains the markdown code.
+   *
+   * @return The markdown channel
+   */
+  public getMarkdownContentChannel(): YText {
+    return this.getText(MARKDOWN_CONTENT_CHANNEL_NAME)
+  }
+
+  /**
+   * Gets the current content of the note as it's currently edited in realtime.
+   *
+   * Please be aware that the return of this method may be very quickly outdated.
+   *
+   * @return The current note content.
+   */
+  public getCurrentContent(): string {
+    return this.getMarkdownContentChannel().toString()
+  }
+}

commons/src/y-doc-sync/y-doc-sync-adapter.test.ts

@@ -0,0 +1,162 @@
+/*
+ * SPDX-FileCopyrightText: 2023 The HedgeDoc developers (see AUTHORS file)
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+import { Message, MessageType } from '../message-transporters/message.js'
+import { InMemoryConnectionMessageTransporter } from './in-memory-connection-message.transporter.js'
+import { RealtimeDoc } from './realtime-doc.js'
+import { YDocSyncClientAdapter } from './y-doc-sync-client-adapter.js'
+import { YDocSyncServerAdapter } from './y-doc-sync-server-adapter.js'
+import { describe, expect, it } from '@jest/globals'
+
+describe('message transporter', () => {
+  it('server client communication', async () => {
+    const docServer: RealtimeDoc = new RealtimeDoc('This is a test note')
+    const docClient1: RealtimeDoc = new RealtimeDoc()
+    const docClient2: RealtimeDoc = new RealtimeDoc()
+
+    const textServer = docServer.getMarkdownContentChannel()
+    const textClient1 = docClient1.getMarkdownContentChannel()
+    const textClient2 = docClient2.getMarkdownContentChannel()
+
+    textServer.observe(() =>
+      console.debug('textServer', new Date(), textServer.toString())
+    )
+    textClient1.observe(() =>
+      console.debug('textClient1', new Date(), textClient1.toString())
+    )
+    textClient2.observe(() =>
+      console.debug('textClient2', new Date(), textClient2.toString())
+    )
+
+    const transporterServerTo1 = new InMemoryConnectionMessageTransporter('s>1')
+    const transporterServerTo2 = new InMemoryConnectionMessageTransporter('s>2')
+    const transporterClient1 = new InMemoryConnectionMessageTransporter('1>s')
+    const transporterClient2 = new InMemoryConnectionMessageTransporter('2>s')
+
+    transporterServerTo1.on(MessageType.NOTE_CONTENT_UPDATE, () =>
+      console.debug('Received NOTE_CONTENT_UPDATE from client 1 to server')
+    )
+    transporterServerTo2.on(MessageType.NOTE_CONTENT_UPDATE, () =>
+      console.debug('Received NOTE_CONTENT_UPDATE from client 2 to server')
+    )
+    transporterClient1.on(MessageType.NOTE_CONTENT_UPDATE, () =>
+      console.debug('Received NOTE_CONTENT_UPDATE from server to client 1')
+    )
+    transporterClient2.on(MessageType.NOTE_CONTENT_UPDATE, () =>
+      console.debug('Received NOTE_CONTENT_UPDATE from server to client 2')
+    )
+    transporterServerTo1.on(MessageType.NOTE_CONTENT_STATE_REQUEST, () =>
+      console.debug('Received NOTE_CONTENT_REQUEST from client 1 to server')
+    )
+    transporterServerTo2.on(MessageType.NOTE_CONTENT_STATE_REQUEST, () =>
+      console.debug('Received NOTE_CONTENT_REQUEST from client 2 to server')
+    )
+    transporterClient1.on(MessageType.NOTE_CONTENT_STATE_REQUEST, () =>
+      console.debug('Received NOTE_CONTENT_REQUEST from server to client 1')
+    )
+    transporterClient2.on(MessageType.NOTE_CONTENT_STATE_REQUEST, () =>
+      console.debug('Received NOTE_CONTENT_REQUEST from server to client 2')
+    )
+    transporterClient1.on('connected', () => console.debug('1>s is connected'))
+    transporterClient2.on('connected', () => console.debug('2>s is connected'))
+    transporterServerTo1.on('connected', () =>
+      console.debug('s>1 is connected')
+    )
+    transporterServerTo2.on('connected', () =>
+      console.debug('s>2 is connected')
+    )
+
+    docServer.on('update', (update: Uint8Array, origin: unknown) => {
+      const message: Message<MessageType.NOTE_CONTENT_UPDATE> = {
+        type: MessageType.NOTE_CONTENT_UPDATE,
+        payload: Array.from(update)
+      }
+      if (origin !== transporterServerTo1) {
+        console.debug('YDoc on Server updated. Sending to Client 1')
+        transporterServerTo1.sendMessage(message)
+      }
+      if (origin !== transporterServerTo2) {
+        console.debug('YDoc on Server updated. Sending to Client 2')
+        transporterServerTo2.sendMessage(message)
+      }
+    })
+    docClient1.on('update', (update: Uint8Array, origin: unknown) => {
+      if (origin !== transporterClient1) {
+        console.debug('YDoc on client 1 updated. Sending to Server')
+      }
+    })
+    docClient2.on('update', (update: Uint8Array, origin: unknown) => {
+      if (origin !== transporterClient2) {
+        console.debug('YDoc on client 2 updated. Sending to Server')
+      }
+    })
+
+    const yDocSyncAdapter1 = new YDocSyncClientAdapter(transporterClient1)
+    yDocSyncAdapter1.setYDoc(docClient1)
+
+    const yDocSyncAdapter2 = new YDocSyncClientAdapter(transporterClient2)
+    yDocSyncAdapter2.setYDoc(docClient2)
+
+    const yDocSyncAdapterServerTo1 = new YDocSyncServerAdapter(
+      transporterServerTo1
+    )
+    yDocSyncAdapterServerTo1.setYDoc(docServer)
+
+    const yDocSyncAdapterServerTo2 = new YDocSyncServerAdapter(
+      transporterServerTo2
+    )
+    yDocSyncAdapterServerTo2.setYDoc(docServer)
+
+    const waitForClient1Sync = new Promise<void>((resolve) => {
+      yDocSyncAdapter1.doAsSoonAsSynced(() => {
+        console.debug('client 1 received the first sync')
+        resolve()
+      })
+    })
+    const waitForClient2Sync = new Promise<void>((resolve) => {
+      yDocSyncAdapter2.doAsSoonAsSynced(() => {
+        console.debug('client 2 received the first sync')
+        resolve()
+      })
+    })
+    const waitForServerTo11Sync = new Promise<void>((resolve) => {
+      yDocSyncAdapterServerTo1.doAsSoonAsSynced(() => {
+        console.debug('server 1 received the first sync')
+        resolve()
+      })
+    })
+    const waitForServerTo21Sync = new Promise<void>((resolve) => {
+      yDocSyncAdapterServerTo2.doAsSoonAsSynced(() => {
+        console.debug('server 2 received the first sync')
+        resolve()
+      })
+    })
+
+    transporterClient1.connect(transporterServerTo1)
+    transporterClient2.connect(transporterServerTo2)
+
+    yDocSyncAdapter1.requestDocumentState()
+    yDocSyncAdapter2.requestDocumentState()
+
+    await Promise.all([
+      waitForClient1Sync,
+      waitForClient2Sync,
+      waitForServerTo11Sync,
+      waitForServerTo21Sync
+    ])
+
+    textClient1.insert(0, 'test2')
+    textClient1.insert(0, 'test3')
+    textClient2.insert(0, 'test4')
+
+    expect(textServer.toString()).toBe('test4test3test2This is a test note')
+    expect(textClient1.toString()).toBe('test4test3test2This is a test note')
+    expect(textClient2.toString()).toBe('test4test3test2This is a test note')
+
+    docServer.destroy()
+    docClient1.destroy()
+    docClient2.destroy()
+  })
+})

commons/src/y-doc-sync/y-doc-sync-adapter.ts

@@ -0,0 +1,148 @@
+/*
+ * SPDX-FileCopyrightText: 2023 The HedgeDoc developers (see AUTHORS file)
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+import { MessageTransporter } from '../message-transporters/message-transporter.js'
+import { Message, MessageType } from '../message-transporters/message.js'
+import { Listener } from 'eventemitter2'
+import { EventEmitter2 } from 'eventemitter2'
+import { applyUpdate, Doc, encodeStateAsUpdate, encodeStateVector } from 'yjs'
+
+type EventMap = Record<'synced' | 'desynced', () => void>
+
+/**
+ * Sends and processes messages that are used to first-synchronize and update a {@link Doc y-doc}.
+ */
+export abstract class YDocSyncAdapter {
+  public readonly eventEmitter = new EventEmitter2<EventMap>()
+
+  protected doc: Doc | undefined
+
+  private destroyYDocUpdateCallback: undefined | (() => void)
+  private destroyEventListenerCallback: undefined | (() => void)
+  private synced = false
+
+  constructor(protected readonly messageTransporter: MessageTransporter) {
+    this.bindDocumentSyncMessageEvents()
+  }
+
+  /**
+   * Executes the given callback as soon as the sync adapter has synchronized the y-doc.
+   * If the y-doc has already been synchronized then the callback is executed immediately.
+   *
+   * @param callback the callback to execute
+   * @return The event listener that waits for the sync event
+   */
+  public doAsSoonAsSynced(callback: () => void): Listener {
+    if (this.isSynced()) {
+      callback()
+    }
+    return this.eventEmitter.on('synced', callback, {
+      objectify: true
+    }) as Listener
+  }
+
+  public getMessageTransporter(): MessageTransporter {
+    return this.messageTransporter
+  }
+
+  public isSynced(): boolean {
+    return this.synced
+  }
+
+  /**
+   * Sets the {@link Doc y-doc} that should be synchronized.
+   *
+   * @param doc the doc to synchronize.
+   */
+  public setYDoc(doc: Doc | undefined): void {
+    this.doc = doc
+
+    this.destroyYDocUpdateCallback?.()
+    if (!doc) {
+      return
+    }
+    const yDocUpdateCallback = this.processDocUpdate.bind(this)
+    doc.on('update', yDocUpdateCallback)
+    this.destroyYDocUpdateCallback = () => doc.off('update', yDocUpdateCallback)
+    this.eventEmitter.emit('desynced')
+  }
+
+  public destroy(): void {
+    this.destroyYDocUpdateCallback?.()
+    this.destroyEventListenerCallback?.()
+  }
+
+  protected bindDocumentSyncMessageEvents(): void {
+    const stateRequestListener = this.messageTransporter.on(
+      MessageType.NOTE_CONTENT_STATE_REQUEST,
+      (payload) => {
+        if (this.doc) {
+          this.messageTransporter.sendMessage({
+            type: MessageType.NOTE_CONTENT_UPDATE,
+            payload: Array.from(
+              encodeStateAsUpdate(this.doc, new Uint8Array(payload.payload))
+            )
+          })
+        }
+      },
+      { objectify: true }
+    ) as Listener
+
+    const disconnectedListener = this.messageTransporter.on(
+      'disconnected',
+      () => {
+        this.synced = false
+        this.eventEmitter.emit('desynced')
+        this.destroy()
+      },
+      { objectify: true }
+    ) as Listener
+
+    const noteContentUpdateListener = this.messageTransporter.on(
+      MessageType.NOTE_CONTENT_UPDATE,
+      (payload) => {
+        if (this.doc) {
+          applyUpdate(this.doc, new Uint8Array(payload.payload), this)
+        }
+      },
+      { objectify: true }
+    ) as Listener
+
+    this.destroyEventListenerCallback = () => {
+      stateRequestListener.off()
+      disconnectedListener.off()
+      noteContentUpdateListener.off()
+    }
+  }
+
+  private processDocUpdate(update: Uint8Array, origin: unknown): void {
+    if (!this.isSynced() || origin === this) {
+      return
+    }
+    const message: Message<MessageType.NOTE_CONTENT_UPDATE> = {
+      type: MessageType.NOTE_CONTENT_UPDATE,
+      payload: Array.from(update)
+    }
+
+    this.messageTransporter.sendMessage(message)
+  }
+
+  protected markAsSynced(): void {
+    if (this.synced) {
+      return
+    }
+    this.synced = true
+    this.eventEmitter.emit('synced')
+  }
+
+  public requestDocumentState(): void {
+    if (this.doc) {
+      this.messageTransporter.sendMessage({
+        type: MessageType.NOTE_CONTENT_STATE_REQUEST,
+        payload: Array.from(encodeStateVector(this.doc))
+      })
+    }
+  }
+}

commons/src/y-doc-sync/y-doc-sync-client-adapter.ts

@@ -0,0 +1,17 @@
+/*
+ * SPDX-FileCopyrightText: 2023 The HedgeDoc developers (see AUTHORS file)
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+import { MessageType } from '../message-transporters/message.js'
+import { YDocSyncAdapter } from './y-doc-sync-adapter.js'
+
+export class YDocSyncClientAdapter extends YDocSyncAdapter {
+  protected bindDocumentSyncMessageEvents() {
+    super.bindDocumentSyncMessageEvents()
+
+    this.messageTransporter.on(MessageType.NOTE_CONTENT_UPDATE, () => {
+      this.markAsSynced()
+    })
+  }
+}

commons/src/y-doc-sync/y-doc-sync-server-adapter.ts

@@ -0,0 +1,14 @@
+/*
+ * SPDX-FileCopyrightText: 2023 The HedgeDoc developers (see AUTHORS file)
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+import { MessageTransporter } from '../message-transporters/message-transporter.js'
+import { YDocSyncAdapter } from './y-doc-sync-adapter.js'
+
+export class YDocSyncServerAdapter extends YDocSyncAdapter {
+  constructor(readonly messageTransporter: MessageTransporter) {
+    super(messageTransporter)
+    this.markAsSynced()
+  }
+}