feat: add logs to pages (#1979)

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

Author: pyphilia

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

@@ -1,3 +1,4 @@
+import { captureException } from '@sentry/node';
 // eslint-disable-next-line @typescript-eslint/ban-ts-comment
 // @ts-expect-error
 import * as decoding from 'lib0/decoding';
@@ -15,6 +16,8 @@ import * as syncProtocol from 'y-protocols/sync';
 // @ts-expect-error
 import * as Y from 'yjs';
 
+import { FastifyBaseLogger } from 'fastify';
+
 import { MESSAGE_AWARENESS_CODE, MESSAGE_SYNC_CODE } from './constants';
 import { PageItemService } from './page.service';
 
@@ -32,15 +35,22 @@ export class WSDoc extends Y.Doc {
   protected name: string;
   protected enableAwareness: boolean;
   protected pageItemService: PageItemService;
+  protected logger: FastifyBaseLogger;
 
-  constructor(pageItemService: PageItemService, name: string, enableAwareness: boolean) {
+  constructor(
+    pageItemService: PageItemService,
+    name: string,
+    enableAwareness: boolean,
+    logger: FastifyBaseLogger,
+  ) {
     super();
     this.name = name;
     this.conns = new Map();
     this.enableAwareness = enableAwareness;
     this.awareness = new awarenessProtocol.Awareness(this);
     this.awareness.setLocalState(null);
     this.pageItemService = pageItemService;
+    this.logger = logger;
   }
 
   protected broadcastUpdate(update: Uint8Array) {
@@ -57,6 +67,7 @@ export class WSDoc extends Y.Doc {
    * @param conn
    */
   addConnection(conn: WebSocket) {
+    this.logger.info('Page', this.name, ': add connection to reach', this.conns.size + 1);
     this.conns.set(conn, new Set());
     // listen and reply to events
     conn.on('message', (message: ArrayBuffer) => {
@@ -74,6 +85,7 @@ export class WSDoc extends Y.Doc {
       const messageType = decoding.readVarUint(decoder);
       switch (messageType) {
         case MESSAGE_SYNC_CODE:
+          this.logger.debug(`Page ${this.name}: receive message`);
           encoding.writeVarUint(encoder, MESSAGE_SYNC_CODE);
           syncProtocol.readSyncMessage(decoder, encoder, this, conn);
 
@@ -86,6 +98,7 @@ export class WSDoc extends Y.Doc {
           break;
         case MESSAGE_AWARENESS_CODE: {
           if (this.enableAwareness) {
+            this.logger.debug(`Page ${this.name}: receive awareness`);
             awarenessProtocol.applyAwarenessUpdate(
               this.awareness,
               decoding.readVarUint8Array(decoder),
@@ -96,7 +109,9 @@ export class WSDoc extends Y.Doc {
         }
       }
     } catch (err) {
-      console.error(err);
+      this.logger.error(err);
+      // send error to sentry
+      captureException(err, { tags: { feature: 'page', pageId: this.name } });
       // close connection because receive message is unsupported
       this.closeConn(conn, 1003);
     }
@@ -117,7 +132,9 @@ export class WSDoc extends Y.Doc {
         err != null && this.closeConn(conn);
       });
     } catch (e) {
-      console.error(e);
+      this.logger.error(e);
+      // send error to sentry
+      captureException(e, { tags: { feature: 'page', pageId: this.name } });
       this.closeConn(conn);
     }
   }
@@ -127,7 +144,7 @@ export class WSDoc extends Y.Doc {
    * @param conn connection to close
    */
   closeConn(conn: WebSocket, code = 1000, reason?: string) {
-    console.debug('Page: close connection', code, reason);
+    this.logger.info(`Page ${this.name}: close connection ${code} ${reason}`);
     if (this.conns.has(conn)) {
       const controlledIds: Set<number> = this.conns.get(conn)!;
       this.conns.delete(conn);

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

@@ -81,7 +81,7 @@ export const pageItemPlugin: FastifyPluginAsyncTypebox = async (fastify) => {
     },
     async (client, req) => {
       client.on('error', fastify.log.error);
-      setupWSConnectionForRead(client, req.params.id, pageItemService);
+      setupWSConnectionForRead(client, req.params.id, pageItemService, fastify.log);
     },
   );
 
@@ -111,7 +111,7 @@ export const pageItemPlugin: FastifyPluginAsyncTypebox = async (fastify) => {
     },
     async (client, req) => {
       client.on('error', fastify.log.error);
-      setupWSConnectionForWriters(client, req.params.id, pageItemService);
+      setupWSConnectionForWriters(client, req.params.id, pageItemService, fastify.log);
     },
   );
 };

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

@@ -17,6 +17,8 @@ import * as syncProtocol from 'y-protocols/sync';
 // @ts-expect-error
 import * as Y from 'yjs';
 
+import { FastifyBaseLogger } from 'fastify';
+
 import { db } from '../../../../drizzle/db';
 import { WSDoc } from './WSDoc';
 import { MESSAGE_AWARENESS_CODE, MESSAGE_SYNC_CODE, PING_TIMEOUT } from './constants';
@@ -40,8 +42,8 @@ class WSSharedDoc extends WSDoc {
     return name + '_shared';
   }
 
-  constructor(pageItemService: PageItemService, name: string) {
-    super(pageItemService, name, true);
+  constructor(pageItemService: PageItemService, name: string, logger: FastifyBaseLogger) {
+    super(pageItemService, name, true, logger);
 
     const awarenessChangeHandler = (
       {
@@ -113,9 +115,9 @@ class WSSharedDoc extends WSDoc {
         this.pageItemService.storeUpdate(db, pageId, update);
       });
     } catch (e) {
-      console.error('An error occured while binding the state:', e);
+      this.logger.error(`Page ${pageId}: An error occured while binding the state: ${e}`);
       // send error to sentry
-      captureException(e);
+      captureException(e, { tags: { feature: 'page', pageId: this.name } });
 
       this.conns.forEach((v, conn) => {
         // close connections for unexpected error
@@ -132,8 +134,8 @@ class WSSharedDoc extends WSDoc {
 class WSReadDoc extends WSDoc {
   private SYNC_ORIGIN = 'sync';
 
-  constructor(pageItemService: PageItemService, name: string) {
-    super(pageItemService, name, false);
+  constructor(pageItemService: PageItemService, name: string, logger: FastifyBaseLogger) {
+    super(pageItemService, name, false, logger);
     this.bindState(name);
 
     // send yjs doc updates to all connections
@@ -150,9 +152,9 @@ class WSReadDoc extends WSDoc {
       const persistedYdoc = await this.pageItemService.getById(db, pageId);
       Y.applyUpdate(this, Y.encodeStateAsUpdate(persistedYdoc), this.SYNC_ORIGIN);
     } catch (e) {
-      console.error('An error occured while binding the state:', e);
+      this.logger.error(`Page ${pageId}: An error occured while binding the state: ${e}`);
       // send error to sentry
-      captureException(e);
+      captureException(e, { tags: { feature: 'page', pageId: this.name } });
       this.conns.forEach((v, conn) => {
         // close connections for unexpected error
         this.closeConn(conn, 1011);
@@ -173,7 +175,7 @@ class WSReadDoc extends WSDoc {
  * @param conn websocket connection
  * @param doc yjs document
  */
-function setupPingPong(conn: WebSocket, doc: WSDoc) {
+function setupPingPong(conn: WebSocket, doc: WSDoc, pageId: string, logger: FastifyBaseLogger) {
   // Check if connection is still alive
   let pongReceived = true;
   const pingInterval = setInterval(() => {
@@ -187,7 +189,7 @@ function setupPingPong(conn: WebSocket, doc: WSDoc) {
       try {
         conn.ping();
       } catch (e) {
-        console.error(e);
+        logger.error(`Page ${pageId}: ${e}`);
         doc.closeConn(conn);
         clearInterval(pingInterval);
       }
@@ -213,30 +215,34 @@ export const setupWSConnectionForWriters = (
   conn: WebSocket,
   pageId: string,
   pageItemService: PageItemService,
+  logger: FastifyBaseLogger,
 ) => {
   conn.binaryType = 'arraybuffer';
   // get doc, initialize if it does not exist yet
   const doc = map.setIfUndefined(docs, pageId, () => {
-    const doc = new WSSharedDoc(pageItemService, pageId);
+    logger.info(`Page ${pageId}: Initializing new doc`);
+    const doc = new WSSharedDoc(pageItemService, pageId, logger);
 
     docs.set(pageId, doc);
 
     return doc;
   });
   doc.addConnection(conn);
 
-  setupPingPong(conn, doc);
+  setupPingPong(conn, doc, pageId, logger);
 
   // put the following in a variables in a block so the interval handlers don't keep in in
   // scope
   {
     // send sync step 1
+    logger.info(`Page ${pageId}: send sync step 1 `);
     const encoder = encoding.createEncoder();
     encoding.writeVarUint(encoder, MESSAGE_SYNC_CODE);
     syncProtocol.writeSyncStep1(encoder, doc);
     doc.send(conn, encoding.toUint8Array(encoder));
 
     // send init awareness
+    logger.info(`Page ${pageId}: send init awareness`);
     const awarenessStates = doc.awareness.getStates();
     if (awarenessStates.size > 0) {
       const encoder = encoding.createEncoder();
@@ -259,23 +265,26 @@ export const setupWSConnectionForRead = (
   conn: WebSocket,
   pageId: string,
   pageItemService: PageItemService,
+  logger: FastifyBaseLogger,
 ) => {
   conn.binaryType = 'arraybuffer';
   // get doc, initialize if it does not exist yet
   const doc = map.setIfUndefined(readDocs, pageId, () => {
-    const doc = new WSReadDoc(pageItemService, pageId);
+    logger.info(`Page ${pageId}: Initializing new doc`);
+    const doc = new WSReadDoc(pageItemService, pageId, logger);
     readDocs.set(pageId, doc);
 
     return doc;
   });
   doc.addConnection(conn);
 
-  setupPingPong(conn, doc);
+  setupPingPong(conn, doc, pageId, logger);
 
   // put the following in a variables in a block so the interval handlers don't keep in in
   // scope
   {
     // send sync step 1
+    logger.info(`Page ${pageId}: send sync step 1`);
     const encoder = encoding.createEncoder();
     encoding.writeVarUint(encoder, MESSAGE_SYNC_CODE);
     syncProtocol.writeSyncStep1(encoder, doc);