switch to nx monorepo

apps/server/src/collaboration/adapter/collab-ws.adapter.ts

@@ -0,0 +1,43 @@
+import { WebSocketServer } from 'ws';
+
+export class CollabWsAdapter {
+  private readonly wss: WebSocketServer;
+
+  constructor() {
+    this.wss = new WebSocketServer({ noServer: true });
+  }
+
+  handleUpgrade(path: string, httpServer) {
+    httpServer.on('upgrade', (request, socket, head) => {
+      try {
+        const baseUrl = 'ws://' + request.headers.host + '/';
+        const pathname = new URL(request.url, baseUrl).pathname;
+
+        if (pathname === path) {
+          this.wss.handleUpgrade(request, socket, head, (ws) => {
+            this.wss.emit('connection', ws, request);
+          });
+        } else if (pathname === '/socket.io/') {
+          return;
+        } else {
+          socket.destroy();
+        }
+      } catch (err) {
+        socket.end('HTTP/1.1 400\r\n' + err.message);
+      }
+    });
+
+    return this.wss;
+  }
+
+  public destroy() {
+    try {
+      this.wss.clients.forEach((client) => {
+        client.terminate();
+      });
+      this.wss.close();
+    } catch (err) {
+      console.error(err);
+    }
+  }
+}

apps/server/src/collaboration/collaboration.gateway.ts

@@ -0,0 +1,34 @@
+import { Server as HocuspocusServer } from '@hocuspocus/server';
+import { IncomingMessage } from 'http';
+import WebSocket from 'ws';
+import { AuthenticationExtension } from './extensions/authentication.extension';
+import { PersistenceExtension } from './extensions/persistence.extension';
+import { Injectable } from '@nestjs/common';
+import { HistoryExtension } from './extensions/history.extension';
+
+@Injectable()
+export class CollaborationGateway {
+  constructor(
+    private authenticationExtension: AuthenticationExtension,
+    private persistenceExtension: PersistenceExtension,
+    private historyExtension: HistoryExtension,
+  ) {}
+
+  private hocuspocus = HocuspocusServer.configure({
+    debounce: 5000,
+    maxDebounce: 10000,
+    extensions: [
+      this.authenticationExtension,
+      this.persistenceExtension,
+      this.historyExtension,
+    ],
+  });
+
+  handleConnection(client: WebSocket, request: IncomingMessage): any {
+    this.hocuspocus.handleConnection(client, request);
+  }
+
+  destroy() {
+    this.hocuspocus.destroy();
+  }
+}

apps/server/src/collaboration/collaboration.module.ts

@@ -0,0 +1,47 @@
+import { Module, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { UserModule } from '../core/user/user.module';
+import { AuthModule } from '../core/auth/auth.module';
+import { AuthenticationExtension } from './extensions/authentication.extension';
+import { PersistenceExtension } from './extensions/persistence.extension';
+import { PageModule } from '../core/page/page.module';
+import { CollaborationGateway } from './collaboration.gateway';
+import { HttpAdapterHost } from '@nestjs/core';
+import { CollabWsAdapter } from './adapter/collab-ws.adapter';
+import { IncomingMessage } from 'http';
+import { WebSocket } from 'ws';
+import { HistoryExtension } from './extensions/history.extension';
+
+@Module({
+  providers: [
+    CollaborationGateway,
+    AuthenticationExtension,
+    PersistenceExtension,
+    HistoryExtension,
+  ],
+  imports: [UserModule, AuthModule, PageModule],
+})
+export class CollaborationModule implements OnModuleInit, OnModuleDestroy {
+  private collabWsAdapter: CollabWsAdapter;
+  private path = '/collaboration';
+
+  constructor(
+    private readonly collaborationGateway: CollaborationGateway,
+    private readonly httpAdapterHost: HttpAdapterHost,
+  ) {}
+
+  onModuleInit() {
+    this.collabWsAdapter = new CollabWsAdapter();
+    const httpServer = this.httpAdapterHost.httpAdapter.getHttpServer();
+
+    const wss = this.collabWsAdapter.handleUpgrade(this.path, httpServer);
+
+    wss.on('connection', (client: WebSocket, request: IncomingMessage) => {
+      this.collaborationGateway.handleConnection(client, request);
+    });
+  }
+
+  onModuleDestroy(): any {
+    this.collaborationGateway.destroy();
+    this.collabWsAdapter.destroy();
+  }
+}

apps/server/src/collaboration/extensions/authentication.extension.ts

@@ -0,0 +1,34 @@
+import { Extension, onAuthenticatePayload } from '@hocuspocus/server';
+import { UserService } from '../../core/user/user.service';
+import { Injectable, UnauthorizedException } from '@nestjs/common';
+import { TokenService } from '../../core/auth/services/token.service';
+
+@Injectable()
+export class AuthenticationExtension implements Extension {
+  constructor(
+    private tokenService: TokenService,
+    private userService: UserService,
+  ) {}
+
+  async onAuthenticate(data: onAuthenticatePayload) {
+    const { documentName, token } = data;
+
+    let jwtPayload = null;
+
+    try {
+      jwtPayload = await this.tokenService.verifyJwt(token);
+    } catch (error) {
+      throw new UnauthorizedException('Could not verify jwt token');
+    }
+
+    const userId = jwtPayload.sub;
+    const user = await this.userService.findById(userId);
+
+    //TODO: Check if the page exists and verify user permissions for page.
+    // if all fails, abort connection
+
+    return {
+      user,
+    };
+  }
+}

apps/server/src/collaboration/extensions/history.extension.ts

@@ -0,0 +1,64 @@
+import {
+  Extension,
+  onChangePayload,
+  onDisconnectPayload,
+} from '@hocuspocus/server';
+import { Injectable } from '@nestjs/common';
+import { PageService } from '../../core/page/services/page.service';
+import { PageHistoryService } from '../../core/page/services/page-history.service';
+
+@Injectable()
+export class HistoryExtension implements Extension {
+  ACTIVE_EDITING_INTERVAL = 10 * 60 * 1000; // 10 minutes
+  historyIntervalMap = new Map<string, NodeJS.Timeout>();
+  lastEditTimeMap = new Map<string, number>();
+
+  constructor(
+    private readonly pageService: PageService,
+    private readonly pageHistoryService: PageHistoryService,
+  ) {}
+
+  async onChange(data: onChangePayload): Promise<void> {
+    const pageId = data.documentName;
+    this.lastEditTimeMap.set(pageId, Date.now());
+
+    if (!this.historyIntervalMap.has(pageId)) {
+      const historyInterval = setInterval(() => {
+        if (this.isActiveEditing(pageId)) {
+          this.recordHistory(pageId);
+        }
+      }, this.ACTIVE_EDITING_INTERVAL);
+      this.historyIntervalMap.set(pageId, historyInterval);
+    }
+  }
+
+  async onDisconnect(data: onDisconnectPayload): Promise<void> {
+    const pageId = data.documentName;
+    if (data.clientsCount === 0) {
+      if (this.historyIntervalMap.has(pageId)) {
+        clearInterval(this.historyIntervalMap.get(pageId));
+        this.historyIntervalMap.delete(pageId);
+        this.lastEditTimeMap.delete(pageId);
+      }
+    }
+  }
+
+  isActiveEditing(pageId: string): boolean {
+    const lastEditTime = this.lastEditTimeMap.get(pageId);
+    if (!lastEditTime) {
+      return false;
+    }
+    return Date.now() - lastEditTime < this.ACTIVE_EDITING_INTERVAL;
+  }
+
+  async recordHistory(pageId: string) {
+    try {
+      const page = await this.pageService.findWithContent(pageId);
+      // Todo: compare if data is the same as the previous version
+      await this.pageHistoryService.saveHistory(page);
+      console.log(`New history created for: ${pageId}`);
+    } catch (err) {
+      console.error('An error occurred saving page history', err);
+    }
+  }
+}

apps/server/src/collaboration/extensions/persistence.extension.ts

@@ -0,0 +1,68 @@
+import {
+  Extension,
+  onLoadDocumentPayload,
+  onStoreDocumentPayload,
+} from '@hocuspocus/server';
+import * as Y from 'yjs';
+import { PageService } from '../../core/page/services/page.service';
+import { Injectable } from '@nestjs/common';
+import { TiptapTransformer } from '@hocuspocus/transformer';
+
+@Injectable()
+export class PersistenceExtension implements Extension {
+  constructor(private readonly pageService: PageService) {}
+
+  async onLoadDocument(data: onLoadDocumentPayload) {
+    const { documentName, document } = data;
+    const pageId = documentName;
+
+    if (!document.isEmpty('default')) {
+      return;
+    }
+
+    const page = await this.pageService.findWithAllFields(pageId);
+
+    if (!page) {
+      console.log('page does not exist.');
+      //TODO: terminate connection if the page does not exist?
+      return;
+    }
+
+    if (page.ydoc) {
+      console.log('ydoc loaded from db');
+
+      const doc = new Y.Doc();
+      const dbState = new Uint8Array(page.ydoc);
+
+      Y.applyUpdate(doc, dbState);
+      return doc;
+    }
+
+    // if no ydoc state in db convert json in page.content to Ydoc.
+    if (page.content) {
+      console.log('converting json to ydoc');
+
+      const ydoc = TiptapTransformer.toYdoc(page.content, 'default');
+      Y.encodeStateAsUpdate(ydoc);
+      return ydoc;
+    }
+
+    console.log('creating fresh ydoc');
+    return new Y.Doc();
+  }
+
+  async onStoreDocument(data: onStoreDocumentPayload) {
+    const { documentName, document, context } = data;
+
+    const pageId = documentName;
+
+    const tiptapJson = TiptapTransformer.fromYdoc(document, 'default');
+    const ydocState = Buffer.from(Y.encodeStateAsUpdate(document));
+
+    try {
+      await this.pageService.updateState(pageId, tiptapJson, ydocState);
+    } catch (err) {
+      console.error(`Failed to update page ${documentName}`);
+    }
+  }
+}