refactor(webui): Integrate existing Mongo SocketIO plugin into new Fastify architecture; Pass logger into plugin instead of entire Fastify Instance (resolves #885). (#1027)

davemarco · web-flow · commit d05863134827 · 2025-07-10T15:20:22.000-04:00
diff --git a/components/webui/server/src/app.ts b/components/webui/server/src/app.ts
@@ -5,7 +5,6 @@ import {
 
 import settings from "../settings.json" with {type: "json"};
 import DbManager from "./plugins/DbManager.js";
-import MongoSocketIoServer from "./plugins/MongoSocketIoServer/index.js";
 import S3Manager from "./plugins/S3Manager.js";
 import exampleRoutes from "./routes/example.js";
 import queryRoutes from "./routes/query.js";
@@ -56,11 +55,6 @@ const FastifyV1App: FastifyPluginAsync<AppPluginOptions> = async (
                 profile: settings.StreamFilesS3Profile,
             }
         );
-        await fastify.register(MongoSocketIoServer, {
-            host: settings.MongoDbHost,
-            port: settings.MongoDbPort,
-            database: settings.MongoDbName,
-        });
     }
 
     // Register the routes
diff --git a/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
@@ -1,9 +1,10 @@
+import {FastifyBaseLogger} from "fastify";
 import type {
     Collection,
     Db,
 } from "mongodb";
 
-import {QueryId} from "../../../../common/index.js";
+import {QueryId} from "../../../../../../../common/index.js";
 import {
     CLIENT_UPDATE_TIMEOUT_MILLIS,
     MongoCustomSocket,
@@ -22,15 +23,19 @@ import {
 class MongoWatcherCollection {
     #collection: Collection;
 
+    #logger: FastifyBaseLogger;
+
     // Active watchers
     #queryIdtoWatcherMap: Map<QueryId, Watcher> = new Map();
 
     /**
      * @param collectionName
+     * @param logger
      * @param mongoDb
      */
-    constructor (collectionName: string, mongoDb: Db) {
+    constructor (collectionName: string, logger: FastifyBaseLogger, mongoDb: Db) {
         this.#collection = mongoDb.collection(collectionName);
+        this.#logger = logger;
     }
 
     /**
@@ -70,7 +75,7 @@ class MongoWatcherCollection {
         const watcher = this.#queryIdtoWatcherMap.get(queryId);
 
         if ("undefined" === typeof watcher) {
-            console.warn(`No watcher found for queryID:${queryId}`);
+            this.#logger.warn(`No watcher found for queryID:${queryId}`);
 
             return false;
         }
@@ -82,7 +87,7 @@ class MongoWatcherCollection {
         }
 
         watcher.changeStream.close().catch((err: unknown) => {
-            console.error(`Error closing watcher for queryID:${queryId}:`, err);
+            this.#logger.error(err, `Error closing watcher for queryID:${queryId}`);
         });
         this.#queryIdtoWatcherMap.delete(queryId);
 
@@ -146,7 +151,7 @@ class MongoWatcherCollection {
             const documents = await this.#collection.find(query, options).toArray();
             return documents;
         } catch (error) {
-            console.error("Error fetching data for query:", error);
+            this.#logger.error(error, "Error fetching data for query");
 
             return [];
         }
@@ -187,7 +192,7 @@ class MongoWatcherCollection {
                     };
 
                     fetchAndEmit().catch((error: unknown) => {
-                        console.error("Error in emitUpdatesWithTimeout:", error);
+                        this.#logger.error(error, "Error in emitUpdatesWithTimeout");
                     });
                     lastEmitTime = Date.now();
                 }, delay);
@@ -196,12 +201,12 @@ class MongoWatcherCollection {
 
         watcher.changeStream.on("change", (change) => {
             if ("invalidate" === change.operationType) {
-                console.log("Change stream received invalidate event for queryID", queryId);
+                this.#logger.info(`Change stream received invalidate event for queryID ${queryId}`);
 
                 return;
             }
             emitUpdateWithTimeout().catch((error: unknown) => {
-                console.error("Error in emitUpdatesWithTimeout:", error);
+                this.#logger.error(error, "Error in emitUpdatesWithTimeout");
             });
         });
     }
diff --git a/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
@@ -7,8 +7,8 @@ import {lookup as dnsLookup} from "node:dns/promises";
 
 import fastifyHttpProxy from "@fastify/http-proxy";
 import {
+    FastifyBaseLogger,
     FastifyInstance,
-    FastifyPluginAsync,
 } from "fastify";
 import fastifyPlugin from "fastify-plugin";
 import {Db} from "mongodb";
@@ -21,18 +21,16 @@ import type {
     Response,
     ServerToClientEvents,
     SocketData,
-} from "../../../../common/index.js";
+} from "../../../../../../../common/index.js";
 import MongoWatcherCollection from "./MongoWatcherCollection.js";
 import {
     ConnectionId,
-    DbOptions,
     MongoCustomSocket,
     QueryParameters,
 } from "./typings.js";
 import {
     getQuery,
     getQueryHash,
-    initializeMongoClient,
     removeItemFromArray,
 } from "./utils.js";
 
@@ -45,7 +43,7 @@ import {
  * names per query, limiting the number of events listeners triggered in the client.
  */
 class MongoSocketIoServer {
-    #fastify: FastifyInstance;
+    #logger: FastifyBaseLogger;
 
     #io: Server<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>;
 
@@ -65,36 +63,36 @@ class MongoSocketIoServer {
     readonly #mongoDb: Db;
 
     /**
-     * Private constructor for MongoSocketIoServer. This is not intended to be invoked publicly.
-     * Instead, use MongoSocketIoServer.create() to create a new instance of the class.
-     *
-     * @param fastify
+     * @param io
+     * @param logger
      * @param mongoDb
      */
-    constructor (fastify: FastifyInstance, mongoDb: Db) {
-        this.#fastify = fastify;
+    private constructor (
+        io: Server<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>,
+        logger: FastifyBaseLogger,
+        mongoDb: Db
+    ) {
+        this.#io = io;
+        this.#logger = logger;
         this.#mongoDb = mongoDb;
-        this.#io = new Server<
-            ClientToServerEvents,
-            ServerToClientEvents,
-            InterServerEvents,
-            SocketData
-        >(fastify.server);
         this.#registerEventListeners();
     }
 
     /**
      * Creates a new MongoSocketIoServer.
      *
      * @param fastify
-     * @param options
      * @return
+     * @throws {Error} When MongoDB database not found
      */
     static async create (
-        fastify: FastifyInstance,
-        options: DbOptions
+        fastify: FastifyInstance
     ): Promise<MongoSocketIoServer> {
-        const mongoDb = await initializeMongoClient(options);
+        const mongoDb = fastify.mongo.db;
+
+        if ("undefined" === typeof mongoDb) {
+            throw new Error("MongoDB database not found");
+        }
 
         // Fastify listens on all resolved addresses for localhost (e.g. `::1` and `127.0.0.1`), but
         // socket.io can only intercept requests on the main server which listens only on the
@@ -124,15 +122,22 @@ class MongoSocketIoServer {
                 JSON.stringify(e)}`);
         }
 
-        return new MongoSocketIoServer(fastify, mongoDb);
+        const io = new Server<
+            ClientToServerEvents,
+            ServerToClientEvents,
+            InterServerEvents,
+            SocketData
+        >(fastify.server);
+
+        return new MongoSocketIoServer(io, fastify.log, mongoDb);
     }
 
     /**
      * Registers event listeners on socket connection.
      */
     #registerEventListeners () {
         this.#io.on("connection", (socket) => {
-            this.#fastify.log.info(`New socket connected with ID:${socket.id}`);
+            this.#logger.info(`New socket connected with ID:${socket.id}`);
             socket.on("disconnect", this.#disconnectListener.bind(this, socket));
             socket.on(
                 "collection::find::subscribe",
@@ -151,7 +156,7 @@ class MongoSocketIoServer {
      * @param socket
      */
     async #disconnectListener (socket: MongoCustomSocket) {
-        this.#fastify.log.info(`Socket:${socket.id} disconnected`);
+        this.#logger.info(`Socket:${socket.id} disconnected`);
         const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socket.id);
 
         if ("undefined" === typeof subscribedQueryIds) {
@@ -163,7 +168,7 @@ class MongoSocketIoServer {
         }
 
         this.#subscribedQueryIdsMap.delete(socket.id);
-        this.#fastify.log.debug(
+        this.#logger.debug(
             "Subscribed queryIDs map" +
             ` ${JSON.stringify(Array.from(this.#subscribedQueryIdsMap.entries()))}`
         );
@@ -232,8 +237,12 @@ class MongoSocketIoServer {
         : MongoWatcherCollection {
         let watcherCollection = this.#collections.get(collectionName);
         if ("undefined" === typeof watcherCollection) {
-            watcherCollection = new MongoWatcherCollection(collectionName, this.#mongoDb);
-            this.#fastify.log.debug(`Initialize Mongo watcher collection:${collectionName}.`);
+            watcherCollection = new MongoWatcherCollection(
+                collectionName,
+                this.#logger,
+                this.#mongoDb
+            );
+            this.#logger.debug(`Initialize Mongo watcher collection:${collectionName}.`);
             this.#collections.set(collectionName, watcherCollection);
         }
 
@@ -258,14 +267,14 @@ class MongoSocketIoServer {
     ): Promise<void> {
         const {collectionName, query, options} = requestArgs;
 
-        this.#fastify.log.debug(
+        this.#logger.debug(
             `Socket:${socket.id} requested query:${JSON.stringify(query)} ` +
             `with options:${JSON.stringify(options)} to collection:${collectionName}`
         );
 
         const hasCollection = await this.#hasCollection(collectionName);
         if (false === hasCollection) {
-            this.#fastify.log.error(`Collection ${collectionName} does not exist in MongoDB`);
+            this.#logger.error(`Collection ${collectionName} does not exist in MongoDB`);
             callback({
                 error: `Collection ${collectionName} does not exist in MongoDB on server`,
             });
@@ -284,7 +293,7 @@ class MongoSocketIoServer {
         callback({data: {queryId, initialDocuments}});
 
         this.#addQueryIdToSubscribedList(queryId, socket.id);
-        this.#fastify.log.info(
+        this.#logger.info(
             `Socket:${socket.id} subscribed to query:${JSON.stringify(query)} ` +
             `with options:${JSON.stringify(options)} ` +
             `on collection:${collectionName} with ID:${queryId}`
@@ -324,7 +333,7 @@ class MongoSocketIoServer {
     #unsubscribe (socket: MongoCustomSocket, queryId: number) {
         const queryHash: string | undefined = this.#queryIdToQueryHashMap.get(queryId);
         if ("undefined" === typeof queryHash) {
-            this.#fastify.log.error(`Query:${queryId} not found in query map`);
+            this.#logger.error(`Query:${queryId} not found in query map`);
 
             return;
         }
@@ -333,26 +342,26 @@ class MongoSocketIoServer {
 
         const collection = this.#collections.get(queryParams.collectionName);
         if ("undefined" === typeof collection) {
-            this.#fastify.log.error(`${queryParams.collectionName} is missing from server`);
+            this.#logger.error(`${queryParams.collectionName} is missing from server`);
 
             return;
         }
 
         const isLastSubscriber = collection.unsubscribe(queryId, socket.id);
-        this.#fastify.log.info(`Socket:${socket.id} unsubscribed from query:${queryId}`);
+        this.#logger.info(`Socket:${socket.id} unsubscribed from query:${queryId}`);
 
         if (isLastSubscriber) {
-            this.#fastify.log.debug(`Query:${queryId} deleted from query map.`);
+            this.#logger.debug(`Query:${queryId} deleted from query map.`);
             this.#queryIdToQueryHashMap.delete(queryId);
         }
 
-        this.#fastify.log.debug(
+        this.#logger.debug(
             "Query ID to query hash map:" +
             ` ${JSON.stringify(Array.from(this.#queryIdToQueryHashMap.entries()))}`
         );
 
         if (false === collection.isReferenced()) {
-            this.#fastify.log.debug(`Collection:${queryParams.collectionName}` +
+            this.#logger.debug(`Collection:${queryParams.collectionName}` +
             " deallocated from server.");
             this.#collections.delete(queryParams.collectionName);
         }
@@ -370,15 +379,15 @@ class MongoSocketIoServer {
         requestArgs: {queryId: number}
     ): Promise<void> {
         const {queryId} = requestArgs;
-        this.#fastify.log.debug(
+        this.#logger.debug(
             `Socket:${socket.id} requested unsubscription to query:${queryId}`
         );
 
         const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socket.id);
         if ("undefined" === typeof subscribedQueryIds ||
             false === subscribedQueryIds.includes(queryId)
         ) {
-            this.#fastify.log.error(`Socket ${socket.id} is not subscribed to ${queryId}`);
+            this.#logger.error(`Socket ${socket.id} is not subscribed to ${queryId}`);
 
             return;
         }
@@ -388,27 +397,24 @@ class MongoSocketIoServer {
 
         removeItemFromArray(subscribedQueryIds, queryId);
 
-        this.#fastify.log.debug(
+        this.#logger.debug(
             `Subscribed queryIDs map ${
                 JSON.stringify(Array.from(this.#subscribedQueryIdsMap.entries()))}`
         );
     }
 }
 
-/**
- * A Fastify plugin callback for setting up the `MongoSocketIoServer`.
- *
- * @param app
- * @param options
- * @param options.database
- * @param options.host
- * @param options.port
- */
-const MongoServerPlugin: FastifyPluginAsync<DbOptions> = async (
-    app: FastifyInstance,
-    options: DbOptions
-) => {
-    await MongoSocketIoServer.create(app, options);
-};
-
-export default fastifyPlugin(MongoServerPlugin);
+declare module "fastify" {
+    export interface FastifyInstance {
+        MongoSocketIoServer: MongoSocketIoServer;
+    }
+}
+
+export default fastifyPlugin(
+    async (fastify) => {
+        fastify.decorate("MongoSocketIoServer", await MongoSocketIoServer.create(fastify));
+    },
+    {
+        name: "MongoSocketIoServer",
+    }
+);
diff --git a/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts
@@ -11,7 +11,7 @@ import {
     InterServerEvents,
     ServerToClientEvents,
     SocketData,
-} from "../../../../common/index.js";
+} from "../../../../../../../common/index.js";
 
 
 /**
diff --git a/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts
