feat: add read ws (#1967)

* feat: add read ws

* test: add test for read doc

* refactor: read doc only accepts messages from corresponding doc

* fix: delete docs only if no one is connected

---------

Co-authored-by: kim <kim.phanhoang@epfl.ch>

Author: pyphilia

src/app.ts

@@ -25,7 +25,9 @@ export default async function (instance: FastifyInstance): Promise<void> {
 
   await instance.register(fws, {
     errorHandler: (error, conn, _req, _reply) => {
-      log.error(`graasp-websockets: an error occured: ${error}\n\tDestroying connection`);
+      log.error(
+        `graasp-websockets: an error occured: ${error.toString()}\n\tDestroying connection`,
+      );
       conn.terminate();
     },
   });

src/services/item/plugins/page/README.md

@@ -10,6 +10,40 @@ The yjs's updates are saved in `page_update` and merged (given an update count t
 
 `page_update` is agnostic of the data `update` it contains.
 
+### Permission Levels
+
+There are 2 websockets connection endpoints:
+
+- `items/pages/:id/ws/read`: For read access, the websocket allows to receive updates for this page. It can be connected to if signed out and the item is public. It does not handle awareness.
+- `items/pages/:id/ws`: For write access, the websocket allows to receive updates and to send updates.
+
+```mermaid
+flowchart TD
+    A[Writer 1] -->|/ws| E{WSSharedDoc}
+    B[Writer 2] -->|/ws| E{WSSharedDoc}
+    B -->|send update, awareness| E
+    A -->|send update, awareness| E
+    C[Reader] -->|/ws/read| D{WSReadDoc}
+    E -->|on: update| D
+    E -->|on: update| A
+    E -->|on: update| B
+    D -->|on: update| C
+
+```
+
+```mermaid
+sequenceDiagram
+    Writer->>WSSharedDoc: /ws
+    WSSharedDoc-->>Writer: first state
+    participant WSReadDoc
+    Reader->>WSReadDoc: /ws/read
+    WSReadDoc-->>Reader: first state
+    Writer->>WSSharedDoc: send update
+    WSSharedDoc->>WSReadDoc: send update
+    WSReadDoc-->>Reader: broadcast update
+
+```
+
 ## Tests
 
 Controller tests use [`y-websocket`](https://github.com/yjs/y-websocket) to connect to the websocket endpoint. This allows to simulate a change in a yjs document to be reflected in the server.

src/services/item/plugins/page/WSDoc.ts

@@ -0,0 +1,136 @@
+// eslint-disable-next-line @typescript-eslint/ban-ts-comment
+// @ts-expect-error
+import * as decoding from 'lib0/decoding';
+// eslint-disable-next-line @typescript-eslint/ban-ts-comment
+// @ts-expect-error
+import * as encoding from 'lib0/encoding';
+import { WebSocket } from 'ws';
+// eslint-disable-next-line @typescript-eslint/ban-ts-comment
+// @ts-expect-error
+import * as awarenessProtocol from 'y-protocols/awareness';
+// eslint-disable-next-line @typescript-eslint/ban-ts-comment
+// @ts-expect-error
+import * as syncProtocol from 'y-protocols/sync';
+// eslint-disable-next-line @typescript-eslint/ban-ts-comment
+// @ts-expect-error
+import * as Y from 'yjs';
+
+import { MESSAGE_AWARENESS_CODE, MESSAGE_SYNC_CODE } from './constants';
+
+const wsReadyStateConnecting = 0;
+const wsReadyStateOpen = 1;
+
+/**
+ * General wrapper of a yjs document
+ * Broadcast updates to attached connections
+ */
+export class WSDoc extends Y.Doc {
+  name: string;
+  conns: Map<WebSocket, Set<number>>;
+  enableAwareness: boolean;
+  awareness: awarenessProtocol.Awareness;
+
+  constructor(name: string, enableAwareness: boolean) {
+    super();
+    this.name = name;
+    this.conns = new Map();
+    this.enableAwareness = enableAwareness;
+    this.awareness = new awarenessProtocol.Awareness(this);
+    this.awareness.setLocalState(null);
+  }
+
+  protected broadcastUpdate(update: Uint8Array) {
+    const encoder = encoding.createEncoder();
+    encoding.writeVarUint(encoder, MESSAGE_SYNC_CODE);
+    syncProtocol.writeUpdate(encoder, update);
+    const message = encoding.toUint8Array(encoder);
+
+    this.conns.forEach((_, conn) => this.send(conn, message));
+  }
+
+  /**
+   * Add a connection to the pool
+   * @param conn
+   */
+  addConnection(conn: WebSocket) {
+    this.conns.set(conn, new Set());
+    // listen and reply to events
+    conn.on('message', (message: ArrayBuffer) => {
+      this.messageListener(conn, new Uint8Array(message));
+    });
+  }
+
+  /*
+   * Get message from other connections at WS connection level
+   */
+  messageListener(conn: WebSocket, message: Uint8Array) {
+    try {
+      const encoder = encoding.createEncoder();
+      const decoder = decoding.createDecoder(message);
+      const messageType = decoding.readVarUint(decoder);
+      switch (messageType) {
+        case MESSAGE_SYNC_CODE:
+          encoding.writeVarUint(encoder, MESSAGE_SYNC_CODE);
+          syncProtocol.readSyncMessage(decoder, encoder, this, conn);
+
+          // If the `encoder` only contains the type of reply message and no
+          // message, there is no need to send the message. When `encoder` only
+          // contains the type of reply, its length is 1.
+          if (encoding.length(encoder) > 1) {
+            this.send(conn, encoding.toUint8Array(encoder));
+          }
+          break;
+        case MESSAGE_AWARENESS_CODE: {
+          if (this.enableAwareness) {
+            awarenessProtocol.applyAwarenessUpdate(
+              this.awareness,
+              decoding.readVarUint8Array(decoder),
+              conn,
+            );
+          }
+          break;
+        }
+      }
+    } catch (err) {
+      console.error(err);
+      // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+      // @ts-expect-error
+      doc.emit('error', [err]);
+    }
+  }
+
+  /**
+   * Send message to given connection
+   * @param conn connection to receive message
+   * @param m message
+   */
+  send(conn: WebSocket, m: Uint8Array) {
+    if (conn.readyState !== wsReadyStateConnecting && conn.readyState !== wsReadyStateOpen) {
+      this.closeConn(conn);
+    }
+    try {
+      conn.send(m, {}, (err) => {
+        // eslint-disable-next-line @typescript-eslint/no-unused-expressions
+        err != null && this.closeConn(conn);
+      });
+    } catch (e) {
+      console.error(e);
+      this.closeConn(conn);
+    }
+  }
+
+  /**
+   * Close given connection
+   * @param conn connection to close
+   */
+  closeConn(conn: WebSocket) {
+    console.debug('Page: close connection');
+    if (this.conns.has(conn)) {
+      const controlledIds: Set<number> = this.conns.get(conn)!;
+      this.conns.delete(conn);
+      awarenessProtocol.removeAwarenessStates(this.awareness, Array.from(controlledIds), null);
+      this.destroy();
+    }
+    conn.close();
+  }
+}

src/services/item/plugins/page/constants.ts

@@ -0,0 +1,4 @@
+export const MESSAGE_SYNC_CODE = 0;
+export const MESSAGE_AWARENESS_CODE = 1;
+
+export const PING_TIMEOUT = 30000;