From 6a4d0b5ed0becc61fc5dfd9bebe5864f9fd78e89 Mon Sep 17 00:00:00 2001
From: Marco <david.marcovitch@yscope.com>
Date: Mon, 23 Jun 2025 18:36:15 +0000
Subject: [PATCH 1/7] latest

---
 components/log-viewer-webui/server/src/app.ts |   7 +-
 .../MongoWatcherCollection.ts                 | 210 ----------
 .../src/plugins/MongoSocketIoServer/index.ts  | 382 ------------------
 .../plugins/MongoSocketIoServer/typings.ts    |  73 ----
 .../src/plugins/MongoSocketIoServer/utils.ts  | 105 -----
 5 files changed, 1 insertion(+), 776 deletions(-)
 delete mode 100644 components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts
 delete mode 100644 components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts
 delete mode 100644 components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts
 delete mode 100644 components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/utils.ts

diff --git a/components/log-viewer-webui/server/src/app.ts b/components/log-viewer-webui/server/src/app.ts
index 1f4a35cafd..e1ee98b9f5 100644
--- a/components/log-viewer-webui/server/src/app.ts
+++ b/components/log-viewer-webui/server/src/app.ts
@@ -5,7 +5,7 @@ import {
 
 import settings from "../settings.json" with {type: "json"};
 import DbManager from "./plugins/DbManager.js";
-import MongoSocketIoServer from "./plugins/MongoSocketIoServer/index.js";
+import MongoSocketIoServer from "./fastify-v2/plugins/app/socket/MongoSocketIoServer/index.js";
 import S3Manager from "./plugins/S3Manager.js";
 import exampleRoutes from "./routes/example.js";
 import queryRoutes from "./routes/query.js";
@@ -56,11 +56,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/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts b/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts
deleted file mode 100644
index 6c2677f202..0000000000
--- a/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts
+++ /dev/null
@@ -1,210 +0,0 @@
-import type {
-    Collection,
-    Db,
-} from "mongodb";
-
-import {QueryId} from "../../../../common/index.js";
-import {
-    CLIENT_UPDATE_TIMEOUT_MILLIS,
-    MongoCustomSocket,
-    QueryParameters,
-    Watcher,
-} from "./typings.js";
-import {
-    convertQueryToChangeStreamFormat,
-    removeItemFromArray,
-} from "./utils.js";
-
-
-/**
- * Provides watchers for MongoDB queries to a specific collection.
- */
-class MongoWatcherCollection {
-    #collection: Collection;
-
-    // Active watchers
-    #queryIdtoWatcherMap: Map<QueryId, Watcher> = new Map();
-
-    /**
-     * @param collectionName
-     * @param mongoDb
-     */
-    constructor (collectionName: string, mongoDb: Db) {
-        this.#collection = mongoDb.collection(collectionName);
-    }
-
-    /**
-     * Checks if the collection is currently being referenced by any clients.
-     *
-     * @return True if the collection is referenced, false otherwise.
-     */
-    isReferenced (): boolean {
-        return 0 !== this.#queryIdtoWatcherMap.size;
-    }
-
-    /**
-     * Checks if a watcher exists for the given query ID.
-     *
-     * @param queryId
-     * @return True if a watcher exists, false otherwise.
-     */
-    hasWatcher (
-        queryId: QueryId,
-    ): boolean {
-        if ("undefined" === typeof this.#queryIdtoWatcherMap.get(queryId)) {
-            return false;
-        }
-
-        return true;
-    }
-
-    /**
-     * Unsubscribes a connection from a watcher. If the watcher has no more subscribers, it closes
-     * the change stream.
-     *
-     * @param queryId
-     * @param connectionId
-     * @return True if connection is last subcriber, false otherwise.
-     */
-    unsubscribe (queryId: number, connectionId: string): boolean {
-        const watcher = this.#queryIdtoWatcherMap.get(queryId);
-
-        if ("undefined" === typeof watcher) {
-            console.warn(`No watcher found for queryID:${queryId}`);
-
-            return false;
-        }
-
-        if (1 < watcher.subscribers.length) {
-            removeItemFromArray(watcher.subscribers, connectionId);
-
-            return false;
-        }
-
-        watcher.changeStream.close().catch((err: unknown) => {
-            console.error(`Error closing watcher for queryID:${queryId}:`, err);
-        });
-        this.#queryIdtoWatcherMap.delete(queryId);
-
-        return true;
-    }
-
-    /**
-     * Adds connection to an existing watcher and joins the the room for the given query ID.
-     *
-     * @param queryId
-     * @param socket
-     */
-    async subscribe (
-        queryId: QueryId,
-        socket: MongoCustomSocket
-    ) {
-        const watcher = this.#queryIdtoWatcherMap.get(queryId);
-
-        if ("undefined" === typeof watcher) {
-            throw new Error(`No watcher found for queryId ${queryId}`);
-        }
-
-        watcher.subscribers.push(socket.id);
-        await socket.join(queryId.toString());
-    }
-
-    /**
-     * Creates a watcher for the given query.
-     *
-     * @param queryParams
-     * @param queryId
-     * @param emitUpdate
-     */
-    createWatcher (
-        queryParams: QueryParameters,
-        queryId: QueryId,
-        emitUpdate: (data: object[]) => void
-    ): void {
-        const watcherQuery = convertQueryToChangeStreamFormat(queryParams.query);
-        const mongoWatcher = this.#collection.watch(
-            [{$match: watcherQuery}],
-            {fullDocument: "updateLookup"}
-        );
-
-        const watcher: Watcher = {changeStream: mongoWatcher, subscribers: []};
-        this.#setupWatcherListener(watcher, queryParams, queryId, emitUpdate);
-        this.#queryIdtoWatcherMap.set(queryId, watcher);
-    }
-
-    /**
-     * Executes a query on the collection and retrieves matching documents.
-     *
-     * @param queryParameters
-     * @return
-     */
-    async find (
-        queryParameters: QueryParameters
-    ): Promise<object[]> {
-        const {query, options} = queryParameters;
-        try {
-            const documents = await this.#collection.find(query, options).toArray();
-            return documents;
-        } catch (error) {
-            console.error("Error fetching data for query:", error);
-
-            return [];
-        }
-    }
-
-    /**
-     * Sets up listener to emit updates to clients on change events.
-     *
-     * @param watcher
-     * @param queryParameters
-     * @param queryId
-     * @param emitUpdate
-     */
-    #setupWatcherListener (
-        watcher: Watcher,
-        queryParameters: QueryParameters,
-        queryId: QueryId,
-        emitUpdate: (data: object[]) => void
-    ) {
-        let lastEmitTime = 0;
-        let emitTimeout: NodeJS.Timeout | null = null;
-
-        const emitUpdateWithTimeout = async () => {
-            const currentTime = Date.now();
-            if (CLIENT_UPDATE_TIMEOUT_MILLIS <= currentTime - lastEmitTime) {
-                lastEmitTime = currentTime;
-                const data = await this.find(queryParameters);
-                emitUpdate(data);
-            } else if (null === emitTimeout) {
-                const delay = CLIENT_UPDATE_TIMEOUT_MILLIS - (currentTime - lastEmitTime);
-
-                emitTimeout = setTimeout(() => {
-                    emitTimeout = null;
-
-                    const fetchAndEmit = async () => {
-                        const data = await this.find(queryParameters);
-                        emitUpdate(data);
-                    };
-
-                    fetchAndEmit().catch((error: unknown) => {
-                        console.error("Error in emitUpdatesWithTimeout:", error);
-                    });
-                    lastEmitTime = Date.now();
-                }, delay);
-            }
-        };
-
-        watcher.changeStream.on("change", (change) => {
-            if ("invalidate" === change.operationType) {
-                console.log("Change stream received invalidate event for queryID", queryId);
-
-                return;
-            }
-            emitUpdateWithTimeout().catch((error: unknown) => {
-                console.error("Error in emitUpdatesWithTimeout:", error);
-            });
-        });
-    }
-}
-
-export default MongoWatcherCollection;
diff --git a/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts b/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts
deleted file mode 100644
index 9fa4c8e0af..0000000000
--- a/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts
+++ /dev/null
@@ -1,382 +0,0 @@
-/* eslint-disable max-lines */
-/* eslint-disable no-warning-comments */
-// TODO: Move listeners to a separate file to reduce lines
-// Reference: https://github.com/socketio/socket.io/blob/main/examples/basic-crud-application/server/lib/todo-management/todo.handlers.ts
-
-import {
-    FastifyInstance,
-    FastifyPluginAsync,
-} from "fastify";
-import fastifyPlugin from "fastify-plugin";
-import {Db} from "mongodb";
-import {Server} from "socket.io";
-
-import type {
-    ClientToServerEvents,
-    InterServerEvents,
-    QueryId,
-    Response,
-    ServerToClientEvents,
-    SocketData,
-} 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";
-
-
-/**
- * Integrates MongoDB with Socket.IO to provide real-time updates for MongoDB queries.
- *
- * TODO: In current implementation, multiple queries in the same collection send updates over
- * one socket with the same event name. A potential improvement would be to use different event
- * names per query, limiting the number of events listeners triggered in the client.
- */
-class MongoSocketIoServer {
-    #fastify: FastifyInstance;
-
-    #io: Server<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>;
-
-    // Collections with active queries.
-    #collections: Map<string, MongoWatcherCollection> = new Map();
-
-    // Mapping of active queries to their hashes.
-    #queryIdToQueryHashMap: Map<QueryId, string> = new Map();
-
-    // Mapping of connection IDs to the query IDs they are subscribed to. A connection can
-    // subscribe to the same queryID multiple times, so the list can contain duplicates.
-    #subscribedQueryIdsMap: Map<ConnectionId, QueryId[]> = new Map();
-
-    // Counter for generating unique query IDs.
-    #queryIdCounter: QueryId = 0;
-
-    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 mongoDb
-     */
-    constructor (fastify: FastifyInstance, mongoDb: Db) {
-        this.#fastify = fastify;
-        this.#mongoDb = mongoDb;
-        this.#io = new Server<
-            ClientToServerEvents,
-            ServerToClientEvents,
-            InterServerEvents,
-            SocketData
-        >(fastify.server);
-        this.#registerEventListeners();
-    }
-
-    /**
-     * Creates a new MongoSocketIoServer.
-     *
-     * @param fastify
-     * @param options
-     * @return
-     */
-    static async create (
-        fastify: FastifyInstance,
-        options: DbOptions
-    ): Promise<MongoSocketIoServer> {
-        const mongoDb = await initializeMongoClient(options);
-        return new MongoSocketIoServer(fastify, mongoDb);
-    }
-
-    /**
-     * Registers event listeners on socket connection.
-     */
-    #registerEventListeners () {
-        this.#io.on("connection", (socket) => {
-            this.#fastify.log.info(`New socket connected with ID:${socket.id}`);
-            socket.on("disconnect", this.#disconnectListener.bind(this, socket));
-            socket.on(
-                "collection::find::subscribe",
-                this.#collectionFindSubscribeListener.bind(this, socket)
-            );
-            socket.on(
-                "collection::find::unsubscribe",
-                this.#collectionFindUnsubscribeListener.bind(this, socket)
-            );
-        });
-    }
-
-    /**
-     * Listener for socket disconnection events.
-     *
-     * @param socket
-     */
-    async #disconnectListener (socket: MongoCustomSocket) {
-        this.#fastify.log.info(`Socket:${socket.id} disconnected`);
-        const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socket.id);
-
-        if ("undefined" === typeof subscribedQueryIds) {
-            return;
-        }
-
-        for (const queryId of subscribedQueryIds) {
-            this.#unsubscribe(socket, queryId);
-        }
-
-        this.#subscribedQueryIdsMap.delete(socket.id);
-        this.#fastify.log.debug(
-            "Subscribed queryIDs map" +
-            ` ${JSON.stringify(Array.from(this.#subscribedQueryIdsMap.entries()))}`
-        );
-    }
-
-    /**
-     * Checks if a collection exists in the MongoDB database.
-     *
-     * @param collectionName
-     * @return Whether the collection exists.
-     */
-    async #hasCollection (collectionName: string): Promise<boolean> {
-        const collections = await this.#mongoDb.listCollections().toArray();
-        return collections.some((collection) => collection.name === collectionName);
-    }
-
-    /**
-     * Adds the query ID to the connection's subscribed query IDs.
-     *
-     * @param queryId
-     * @param socketId
-     */
-    #addQueryIdToSubscribedList (queryId: QueryId, socketId: ConnectionId) {
-        const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socketId);
-        if ("undefined" === typeof subscribedQueryIds) {
-            this.#subscribedQueryIdsMap.set(socketId, [queryId]);
-
-            return;
-        }
-
-        this.#subscribedQueryIdsMap.set(socketId, [...subscribedQueryIds,
-            queryId]);
-    }
-
-    /**
-     * Gets query ID based on query parameters. If not found, creates a new ID.
-     *
-     * @param queryParams
-     * @return the query ID.
-     */
-    #getQueryId (queryParams: QueryParameters): number {
-        const queryHash = getQueryHash(queryParams);
-        for (const [queryId, hash] of this.#queryIdToQueryHashMap.entries()) {
-            if (hash === queryHash) {
-                return queryId;
-            }
-        }
-        const queryId = this.#queryIdCounter;
-        this.#queryIdToQueryHashMap.set(queryId, queryHash);
-
-        // JS is single threaded and ++ is atomic, so we can safely increment the global counter.
-        this.#queryIdCounter++;
-
-        return queryId;
-    }
-
-    /**
-     * Gets an existing watcher collection or creates a new one if it doesn't exist.
-     *
-     * @param collectionName
-     * @return The watcher collection instance.
-     */
-    #getOrCreateWatcherCollection (
-        collectionName: string
-    )
-        : 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}.`);
-            this.#collections.set(collectionName, watcherCollection);
-        }
-
-        return watcherCollection;
-    }
-
-    /**
-     * Listener for subscribing to a find query. The client will receive updates whenever
-     * the query results change.
-     *
-     * @param socket
-     * @param requestArgs
-     * @param requestArgs.query
-     * @param requestArgs.options
-     * @param requestArgs.collectionName
-     * @param callback
-     */
-    async #collectionFindSubscribeListener (
-        socket: MongoCustomSocket,
-        requestArgs: {collectionName: string; query: object; options: object},
-        callback: (res: Response<{queryId: number; initialDocuments: object[]}>) => void
-    ): Promise<void> {
-        const {collectionName, query, options} = requestArgs;
-
-        this.#fastify.log.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`);
-            callback({
-                error: `Collection ${collectionName} does not exist in MongoDB on server`,
-            });
-
-            return;
-        }
-
-        const watcherCollection = this.#getOrCreateWatcherCollection(collectionName);
-
-        const queryParameters: QueryParameters = {collectionName, query, options};
-        const queryId = this.#getQueryId(queryParameters);
-
-        await this.#subscribeToQuery(watcherCollection, queryParameters, queryId, socket);
-
-        const initialDocuments = await watcherCollection.find(queryParameters);
-        callback({data: {queryId, initialDocuments}});
-
-        this.#addQueryIdToSubscribedList(queryId, socket.id);
-        this.#fastify.log.info(
-            `Socket:${socket.id} subscribed to query:${JSON.stringify(query)} ` +
-            `with options:${JSON.stringify(options)} ` +
-            `on collection:${collectionName} with ID:${queryId}`
-        );
-    }
-
-    /**
-     * Subscribes to query updates.
-     *
-     * @param watcherCollection
-     * @param queryParameters
-     * @param queryId
-     * @param socket
-     */
-    async #subscribeToQuery (
-        watcherCollection: MongoWatcherCollection,
-        queryParameters: QueryParameters,
-        queryId: QueryId,
-        socket: MongoCustomSocket
-    ): Promise<void> {
-        if (false === watcherCollection.hasWatcher(queryId)) {
-            const emitUpdate = (data: object[]) => {
-                this.#io.to(`${queryId}`).emit("collection::find::update", {queryId, data});
-            };
-
-            watcherCollection.createWatcher(queryParameters, queryId, emitUpdate);
-        }
-        await watcherCollection.subscribe(queryId, socket);
-    }
-
-    /**
-     * Unsubscribes from a query.
-     *
-     * @param socket
-     * @param queryId
-     */
-    #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`);
-
-            return;
-        }
-
-        const queryParams: QueryParameters = getQuery(queryHash);
-
-        const collection = this.#collections.get(queryParams.collectionName);
-        if ("undefined" === typeof collection) {
-            this.#fastify.log.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}`);
-
-        if (isLastSubscriber) {
-            this.#fastify.log.debug(`Query:${queryId} deleted from query map.`);
-            this.#queryIdToQueryHashMap.delete(queryId);
-        }
-
-        this.#fastify.log.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}` +
-            " deallocated from server.");
-            this.#collections.delete(queryParams.collectionName);
-        }
-    }
-
-    /**
-     * Listener for unsubscribing from a find query.
-     *
-     * @param socket
-     * @param requestArgs
-     * @param requestArgs.queryId
-     */
-    async #collectionFindUnsubscribeListener (
-        socket: MongoCustomSocket,
-        requestArgs: {queryId: number}
-    ): Promise<void> {
-        const {queryId} = requestArgs;
-        this.#fastify.log.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}`);
-
-            return;
-        }
-
-        this.#unsubscribe(socket, queryId);
-        await socket.leave(queryId.toString());
-
-        removeItemFromArray(subscribedQueryIds, queryId);
-
-        this.#fastify.log.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);
diff --git a/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts b/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts
deleted file mode 100644
index 2c1b1e0a96..0000000000
--- a/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts
+++ /dev/null
@@ -1,73 +0,0 @@
-import {
-    ChangeStream,
-    Document,
-    Filter,
-    FindOptions,
-} from "mongodb";
-import {Socket} from "socket.io";
-
-import {
-    ClientToServerEvents,
-    InterServerEvents,
-    ServerToClientEvents,
-    SocketData,
-} from "../../../../common/index.js";
-
-
-/**
- * Custom socket type for Mongo Socket IO server.
- */
-type MongoCustomSocket = Socket<
-    ClientToServerEvents,
-    ServerToClientEvents,
-    InterServerEvents,
-    SocketData
->;
-
-/**
- * Unique ID to represent each socket connection.
- */
-type ConnectionId = string;
-
-
-/**
- * Parameters for MongoDB queries.
- */
-interface QueryParameters {
-    collectionName: string;
-    query: Filter<Document>;
-    options: FindOptions;
-}
-
-/**
- * Options to connect to MongoDB database.
- */
-interface DbOptions {
-    // Name of database.
-    database: string;
-    host: string;
-    port: number;
-}
-
-/**
- * Timeout for emitting updates to the client.
- */
-const CLIENT_UPDATE_TIMEOUT_MILLIS = 500;
-
-/**
- * MongoDB change stream for a query, and a list of subscribed connections. Subscribed connections
- * can include duplicates if the same connection subscribes to the same query multiple times.
- */
-interface Watcher {
-    changeStream: ChangeStream;
-    subscribers: ConnectionId[];
-}
-
-export {
-    CLIENT_UPDATE_TIMEOUT_MILLIS,
-    ConnectionId,
-    DbOptions,
-    MongoCustomSocket,
-    QueryParameters,
-    Watcher,
-};
diff --git a/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/utils.ts b/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/utils.ts
deleted file mode 100644
index acdb8dd9ea..0000000000
--- a/components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/utils.ts
+++ /dev/null
@@ -1,105 +0,0 @@
-import {
-    Db,
-    Document,
-    Filter,
-    MongoClient,
-} from "mongodb";
-
-import {
-    DbOptions,
-    QueryParameters,
-} from "./typings.js";
-
-
-/**
- * Modifies query so that it can be used to filter change stream update events. Update events in
- * MongoDB return the document in the "fullDocument" field, so function prepends "fullDocument" to
- * each key in the query.
- *
- * Reference: https://www.mongodb.com/docs/manual/reference/change-events/update/#mongodb-data-update
- *
- * @param query
- * @return Modified query.
- */
-const convertQueryToChangeStreamFormat = (
-    query: Filter<Document>
-): Filter<Document> => {
-    const changeStreamQuery: Filter<Document> = {};
-    for (const key in query) {
-        if (Object.hasOwn(query, key)) {
-            changeStreamQuery[`fullDocument.${key}`] = query[key] as unknown;
-        }
-    }
-
-    return changeStreamQuery;
-};
-
-/**
- * Generates a unique hash for a given query parameters.
- *
- * @param queryParams
- * @return Unique hash for query parameters.
- */
-const getQueryHash = (
-    queryParams: QueryParameters
-): string => {
-    return JSON.stringify(queryParams);
-};
-
-/**
- * Recovers query parameters from the hash.
- *
- * @param queryHash
- * @return
- */
-const getQuery = (
-    queryHash: string
-): QueryParameters => {
-    const parsedValue: unknown = JSON.parse(queryHash);
-    return parsedValue as QueryParameters;
-};
-
-/**
- * Initializes a MongoDB client.
- *
- * @param options
- * @return
- * @throws {Error} If there is a MongoDB connection error.
- */
-const initializeMongoClient = async (
-    options: DbOptions
-): Promise<Db> => {
-    const mongoUri = `mongodb://${options.host}:${options.port}`;
-    const mongoClient = new MongoClient(mongoUri);
-    try {
-        await mongoClient.connect();
-
-        return mongoClient.db(options.database);
-    } catch (e) {
-        throw new Error("MongoDB connection error", {cause: e as Error});
-    }
-};
-
-/**
- * Removes the first instance of item from an array, if it exists.
- *
- * @param array
- * @param item
- */
-const removeItemFromArray = <T>(
-    array: T[],
-    item: T
-): void => {
-    const index = array.indexOf(item);
-    if (-1 !== index) {
-        array.splice(index, 1);
-    }
-};
-
-export {
-    convertQueryToChangeStreamFormat,
-    getQuery,
-    getQueryHash,
-    initializeMongoClient,
-    removeItemFromArray,
-};

From f738fdf18483f80fc190f14b82b1522ef7d3416c Mon Sep 17 00:00:00 2001
From: Marco <david.marcovitch@yscope.com>
Date: Mon, 23 Jun 2025 18:37:44 +0000
Subject: [PATCH 2/7] latest

---
 .../MongoWatcherCollection.ts                 | 215 ++++++++++
 .../app/socket/MongoSocketIoServer/index.ts   | 387 ++++++++++++++++++
 .../app/socket/MongoSocketIoServer/typings.ts |  73 ++++
 .../app/socket/MongoSocketIoServer/utils.ts   |  80 ++++
 4 files changed, 755 insertions(+)
 create mode 100644 components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
 create mode 100644 components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
 create mode 100644 components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts
 create mode 100644 components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts

diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
new file mode 100644
index 0000000000..d76515ef82
--- /dev/null
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
@@ -0,0 +1,215 @@
+import type {
+    Collection,
+    Db,
+} from "mongodb";
+import {FastifyBaseLogger} from "fastify";
+
+import {QueryId} from "../../../../../../../common/index.js";
+import {
+    CLIENT_UPDATE_TIMEOUT_MILLIS,
+    MongoCustomSocket,
+    QueryParameters,
+    Watcher,
+} from "./typings.js";
+import {
+    convertQueryToChangeStreamFormat,
+    removeItemFromArray,
+} from "./utils.js";
+
+
+/**
+ * Provides watchers for MongoDB queries to a specific collection.
+ */
+class MongoWatcherCollection {
+    #collection: Collection;
+
+    #logger: FastifyBaseLogger;
+
+    // Active watchers
+    #queryIdtoWatcherMap: Map<QueryId, Watcher> = new Map();
+
+    /**
+     * @param collectionName
+     * @param mongoDb
+     * @param logger
+     */
+    constructor (collectionName: string, mongoDb: Db, logger: FastifyBaseLogger) {
+        this.#collection = mongoDb.collection(collectionName);
+        this.#logger = logger;
+    }
+
+    /**
+     * Checks if the collection is currently being referenced by any clients.
+     *
+     * @return True if the collection is referenced, false otherwise.
+     */
+    isReferenced (): boolean {
+        return 0 !== this.#queryIdtoWatcherMap.size;
+    }
+
+    /**
+     * Checks if a watcher exists for the given query ID.
+     *
+     * @param queryId
+     * @return True if a watcher exists, false otherwise.
+     */
+    hasWatcher (
+        queryId: QueryId,
+    ): boolean {
+        if ("undefined" === typeof this.#queryIdtoWatcherMap.get(queryId)) {
+            return false;
+        }
+
+        return true;
+    }
+
+    /**
+     * Unsubscribes a connection from a watcher. If the watcher has no more subscribers, it closes
+     * the change stream.
+     *
+     * @param queryId
+     * @param connectionId
+     * @return True if connection is last subcriber, false otherwise.
+     */
+    unsubscribe (queryId: number, connectionId: string): boolean {
+        const watcher = this.#queryIdtoWatcherMap.get(queryId);
+
+        if ("undefined" === typeof watcher) {
+            this.#logger.warn(`No watcher found for queryID:${queryId}`);
+
+            return false;
+        }
+
+        if (1 < watcher.subscribers.length) {
+            removeItemFromArray(watcher.subscribers, connectionId);
+
+            return false;
+        }
+
+        watcher.changeStream.close().catch((err: unknown) => {
+            this.#logger.error(err, `Error closing watcher for queryID:${queryId}`);
+        });
+        this.#queryIdtoWatcherMap.delete(queryId);
+
+        return true;
+    }
+
+    /**
+     * Adds connection to an existing watcher and joins the the room for the given query ID.
+     *
+     * @param queryId
+     * @param socket
+     */
+    async subscribe (
+        queryId: QueryId,
+        socket: MongoCustomSocket
+    ) {
+        const watcher = this.#queryIdtoWatcherMap.get(queryId);
+
+        if ("undefined" === typeof watcher) {
+            throw new Error(`No watcher found for queryId ${queryId}`);
+        }
+
+        watcher.subscribers.push(socket.id);
+        await socket.join(queryId.toString());
+    }
+
+    /**
+     * Creates a watcher for the given query.
+     *
+     * @param queryParams
+     * @param queryId
+     * @param emitUpdate
+     */
+    createWatcher (
+        queryParams: QueryParameters,
+        queryId: QueryId,
+        emitUpdate: (data: object[]) => void
+    ): void {
+        const watcherQuery = convertQueryToChangeStreamFormat(queryParams.query);
+        const mongoWatcher = this.#collection.watch(
+            [{$match: watcherQuery}],
+            {fullDocument: "updateLookup"}
+        );
+
+        const watcher: Watcher = {changeStream: mongoWatcher, subscribers: []};
+        this.#setupWatcherListener(watcher, queryParams, queryId, emitUpdate);
+        this.#queryIdtoWatcherMap.set(queryId, watcher);
+    }
+
+    /**
+     * Executes a query on the collection and retrieves matching documents.
+     *
+     * @param queryParameters
+     * @return
+     */
+    async find (
+        queryParameters: QueryParameters
+    ): Promise<object[]> {
+        const {query, options} = queryParameters;
+        try {
+            const documents = await this.#collection.find(query, options).toArray();
+            return documents;
+        } catch (error) {
+            this.#logger.error(error, "Error fetching data for query");
+
+            return [];
+        }
+    }
+
+    /**
+     * Sets up listener to emit updates to clients on change events.
+     *
+     * @param watcher
+     * @param queryParameters
+     * @param queryId
+     * @param emitUpdate
+     */
+    #setupWatcherListener (
+        watcher: Watcher,
+        queryParameters: QueryParameters,
+        queryId: QueryId,
+        emitUpdate: (data: object[]) => void
+    ) {
+        let lastEmitTime = 0;
+        let emitTimeout: NodeJS.Timeout | null = null;
+
+        const emitUpdateWithTimeout = async () => {
+            const currentTime = Date.now();
+            if (CLIENT_UPDATE_TIMEOUT_MILLIS <= currentTime - lastEmitTime) {
+                lastEmitTime = currentTime;
+                const data = await this.find(queryParameters);
+                emitUpdate(data);
+            } else if (null === emitTimeout) {
+                const delay = CLIENT_UPDATE_TIMEOUT_MILLIS - (currentTime - lastEmitTime);
+
+                emitTimeout = setTimeout(() => {
+                    emitTimeout = null;
+
+                    const fetchAndEmit = async () => {
+                        const data = await this.find(queryParameters);
+                        emitUpdate(data);
+                    };
+
+                    fetchAndEmit().catch((error: unknown) => {
+                        this.#logger.error(error, "Error in emitUpdatesWithTimeout");
+                    });
+                    lastEmitTime = Date.now();
+                }, delay);
+            }
+        };
+
+        watcher.changeStream.on("change", (change) => {
+            if ("invalidate" === change.operationType) {
+                this.#logger.info(`Change stream received invalidate event for queryID ${queryId}`);
+
+                return;
+            }
+            emitUpdateWithTimeout().catch((error: unknown) => {
+                this.#logger.error(error, "Error in emitUpdatesWithTimeout");
+            });
+        });
+    }
+}
+
+export default MongoWatcherCollection;
diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
new file mode 100644
index 0000000000..84791f8f7f
--- /dev/null
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
@@ -0,0 +1,387 @@
+/* eslint-disable max-lines */
+/* eslint-disable no-warning-comments */
+// TODO: Move listeners to a separate file to reduce lines
+// Reference: https://github.com/socketio/socket.io/blob/main/examples/basic-crud-application/server/lib/todo-management/todo.handlers.ts
+
+import {
+    FastifyInstance,
+} from "fastify";
+import fastifyPlugin from "fastify-plugin";
+import {Db} from "mongodb";
+import {Server} from "socket.io";
+
+import type {
+    ClientToServerEvents,
+    InterServerEvents,
+    QueryId,
+    Response,
+    ServerToClientEvents,
+    SocketData,
+} from "../../../../../../../common/index.js";
+import MongoWatcherCollection from "./MongoWatcherCollection.js";
+import {
+    ConnectionId,
+    MongoCustomSocket,
+    QueryParameters,
+} from "./typings.js";
+import {
+    getQuery,
+    getQueryHash,
+    removeItemFromArray,
+} from "./utils.js";
+import {FastifyBaseLogger} from "fastify";
+
+
+/**
+ * Integrates MongoDB with Socket.IO to provide real-time updates for MongoDB queries.
+ *
+ * TODO: In current implementation, multiple queries in the same collection send updates over
+ * one socket with the same event name. A potential improvement would be to use different event
+ * names per query, limiting the number of events listeners triggered in the client.
+ */
+class MongoSocketIoServer {
+    #logger: FastifyBaseLogger;
+
+    #io: Server<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>;
+
+    // Collections with active queries.
+    #collections: Map<string, MongoWatcherCollection> = new Map();
+
+    // Mapping of active queries to their hashes.
+    #queryIdToQueryHashMap: Map<QueryId, string> = new Map();
+
+    // Mapping of connection IDs to the query IDs they are subscribed to. A connection can
+    // subscribe to the same queryID multiple times, so the list can contain duplicates.
+    #subscribedQueryIdsMap: Map<ConnectionId, QueryId[]> = new Map();
+
+    // Counter for generating unique query IDs.
+    #queryIdCounter: QueryId = 0;
+
+    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 io
+     * @param mongoDb
+     * @param logger
+     */
+    constructor (
+        io: Server<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>,
+        mongoDb: Db,
+        logger: FastifyBaseLogger
+    ) {
+        this.#logger = logger;
+        this.#mongoDb = mongoDb;
+        this.#io = io;
+        this.#registerEventListeners();
+    }
+
+    /**
+     * Creates a new MongoSocketIoServer.
+     *
+     * @param fastify
+     * @return
+     */
+    static create (
+        fastify: FastifyInstance
+    ): MongoSocketIoServer {
+        const mongoDb = fastify.mongo.db;
+
+        if ("undefined" === typeof mongoDb) {
+            throw new Error("MongoDB database not found");
+        }
+
+        const io = new Server<
+            ClientToServerEvents,
+            ServerToClientEvents,
+            InterServerEvents,
+            SocketData
+        >(fastify.server);
+
+        return new MongoSocketIoServer(io, mongoDb, fastify.log);
+    }
+
+    /**
+     * Registers event listeners on socket connection.
+     */
+    #registerEventListeners () {
+        this.#io.on("connection", (socket) => {
+            this.#logger.info(`New socket connected with ID:${socket.id}`);
+            socket.on("disconnect", this.#disconnectListener.bind(this, socket));
+            socket.on(
+                "collection::find::subscribe",
+                this.#collectionFindSubscribeListener.bind(this, socket)
+            );
+            socket.on(
+                "collection::find::unsubscribe",
+                this.#collectionFindUnsubscribeListener.bind(this, socket)
+            );
+        });
+    }
+
+    /**
+     * Listener for socket disconnection events.
+     *
+     * @param socket
+     */
+    async #disconnectListener (socket: MongoCustomSocket) {
+        this.#logger.info(`Socket:${socket.id} disconnected`);
+        const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socket.id);
+
+        if ("undefined" === typeof subscribedQueryIds) {
+            return;
+        }
+
+        for (const queryId of subscribedQueryIds) {
+            this.#unsubscribe(socket, queryId);
+        }
+
+        this.#subscribedQueryIdsMap.delete(socket.id);
+        this.#logger.debug(
+            "Subscribed queryIDs map" +
+            ` ${JSON.stringify(Array.from(this.#subscribedQueryIdsMap.entries()))}`
+        );
+    }
+
+    /**
+     * Checks if a collection exists in the MongoDB database.
+     *
+     * @param collectionName
+     * @return Whether the collection exists.
+     */
+    async #hasCollection (collectionName: string): Promise<boolean> {
+        const collections = await this.#mongoDb.listCollections().toArray();
+        return collections.some((collection) => collection.name === collectionName);
+    }
+
+    /**
+     * Adds the query ID to the connection's subscribed query IDs.
+     *
+     * @param queryId
+     * @param socketId
+     */
+    #addQueryIdToSubscribedList (queryId: QueryId, socketId: ConnectionId) {
+        const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socketId);
+        if ("undefined" === typeof subscribedQueryIds) {
+            this.#subscribedQueryIdsMap.set(socketId, [queryId]);
+
+            return;
+        }
+
+        this.#subscribedQueryIdsMap.set(socketId, [...subscribedQueryIds,
+            queryId]);
+    }
+
+    /**
+     * Gets query ID based on query parameters. If not found, creates a new ID.
+     *
+     * @param queryParams
+     * @return the query ID.
+     */
+    #getQueryId (queryParams: QueryParameters): number {
+        const queryHash = getQueryHash(queryParams);
+        for (const [queryId, hash] of this.#queryIdToQueryHashMap.entries()) {
+            if (hash === queryHash) {
+                return queryId;
+            }
+        }
+        const queryId = this.#queryIdCounter;
+        this.#queryIdToQueryHashMap.set(queryId, queryHash);
+
+        // JS is single threaded and ++ is atomic, so we can safely increment the global counter.
+        this.#queryIdCounter++;
+
+        return queryId;
+    }
+
+    /**
+     * Gets an existing watcher collection or creates a new one if it doesn't exist.
+     *
+     * @param collectionName
+     * @return The watcher collection instance.
+     */
+    #getOrCreateWatcherCollection (
+        collectionName: string
+    )
+        : MongoWatcherCollection {
+        let watcherCollection = this.#collections.get(collectionName);
+        if ("undefined" === typeof watcherCollection) {
+            watcherCollection = new MongoWatcherCollection(collectionName, this.#mongoDb, this.#logger);
+            this.#logger.debug(`Initialize Mongo watcher collection:${collectionName}.`);
+            this.#collections.set(collectionName, watcherCollection);
+        }
+
+        return watcherCollection;
+    }
+
+    /**
+     * Listener for subscribing to a find query. The client will receive updates whenever
+     * the query results change.
+     *
+     * @param socket
+     * @param requestArgs
+     * @param requestArgs.query
+     * @param requestArgs.options
+     * @param requestArgs.collectionName
+     * @param callback
+     */
+    async #collectionFindSubscribeListener (
+        socket: MongoCustomSocket,
+        requestArgs: {collectionName: string; query: object; options: object},
+        callback: (res: Response<{queryId: number; initialDocuments: object[]}>) => void
+    ): Promise<void> {
+        const {collectionName, query, options} = requestArgs;
+
+        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.#logger.error(`Collection ${collectionName} does not exist in MongoDB`);
+            callback({
+                error: `Collection ${collectionName} does not exist in MongoDB on server`,
+            });
+
+            return;
+        }
+
+        const watcherCollection = this.#getOrCreateWatcherCollection(collectionName);
+
+        const queryParameters: QueryParameters = {collectionName, query, options};
+        const queryId = this.#getQueryId(queryParameters);
+
+        await this.#subscribeToQuery(watcherCollection, queryParameters, queryId, socket);
+
+        const initialDocuments = await watcherCollection.find(queryParameters);
+        callback({data: {queryId, initialDocuments}});
+
+        this.#addQueryIdToSubscribedList(queryId, socket.id);
+        this.#logger.info(
+            `Socket:${socket.id} subscribed to query:${JSON.stringify(query)} ` +
+            `with options:${JSON.stringify(options)} ` +
+            `on collection:${collectionName} with ID:${queryId}`
+        );
+    }
+
+    /**
+     * Subscribes to query updates.
+     *
+     * @param watcherCollection
+     * @param queryParameters
+     * @param queryId
+     * @param socket
+     */
+    async #subscribeToQuery (
+        watcherCollection: MongoWatcherCollection,
+        queryParameters: QueryParameters,
+        queryId: QueryId,
+        socket: MongoCustomSocket
+    ): Promise<void> {
+        if (false === watcherCollection.hasWatcher(queryId)) {
+            const emitUpdate = (data: object[]) => {
+                this.#io.to(`${queryId}`).emit("collection::find::update", {queryId, data});
+            };
+
+            watcherCollection.createWatcher(queryParameters, queryId, emitUpdate);
+        }
+        await watcherCollection.subscribe(queryId, socket);
+    }
+
+    /**
+     * Unsubscribes from a query.
+     *
+     * @param socket
+     * @param queryId
+     */
+    #unsubscribe (socket: MongoCustomSocket, queryId: number) {
+        const queryHash: string | undefined = this.#queryIdToQueryHashMap.get(queryId);
+        if ("undefined" === typeof queryHash) {
+            this.#logger.error(`Query:${queryId} not found in query map`);
+
+            return;
+        }
+
+        const queryParams: QueryParameters = getQuery(queryHash);
+
+        const collection = this.#collections.get(queryParams.collectionName);
+        if ("undefined" === typeof collection) {
+            this.#logger.error(`${queryParams.collectionName} is missing from server`);
+
+            return;
+        }
+
+        const isLastSubscriber = collection.unsubscribe(queryId, socket.id);
+        this.#logger.info(`Socket:${socket.id} unsubscribed from query:${queryId}`);
+
+        if (isLastSubscriber) {
+            this.#logger.debug(`Query:${queryId} deleted from query map.`);
+            this.#queryIdToQueryHashMap.delete(queryId);
+        }
+
+        this.#logger.debug(
+            "Query ID to query hash map:" +
+            ` ${JSON.stringify(Array.from(this.#queryIdToQueryHashMap.entries()))}`
+        );
+
+        if (false === collection.isReferenced()) {
+            this.#logger.debug(`Collection:${queryParams.collectionName}` +
+            " deallocated from server.");
+            this.#collections.delete(queryParams.collectionName);
+        }
+    }
+
+    /**
+     * Listener for unsubscribing from a find query.
+     *
+     * @param socket
+     * @param requestArgs
+     * @param requestArgs.queryId
+     */
+    async #collectionFindUnsubscribeListener (
+        socket: MongoCustomSocket,
+        requestArgs: {queryId: number}
+    ): Promise<void> {
+        const {queryId} = requestArgs;
+        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.#logger.error(`Socket ${socket.id} is not subscribed to ${queryId}`);
+
+            return;
+        }
+
+        this.#unsubscribe(socket, queryId);
+        await socket.leave(queryId.toString());
+
+        removeItemFromArray(subscribedQueryIds, queryId);
+
+        this.#logger.debug(
+            `Subscribed queryIDs map ${
+                JSON.stringify(Array.from(this.#subscribedQueryIdsMap.entries()))}`
+        );
+    }
+}
+
+declare module "fastify" {
+    export interface FastifyInstance {
+        MongoSocketIoServer: MongoSocketIoServer;
+    }
+}
+
+export default fastifyPlugin(
+    (fastify) => {
+        fastify.decorate("MongoSocketIoServer", MongoSocketIoServer.create(fastify));
+    },
+    {
+        name: "MongoSocketIoServer",
+    }
+);
diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts
new file mode 100644
index 0000000000..13d8635529
--- /dev/null
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts
@@ -0,0 +1,73 @@
+import {
+    ChangeStream,
+    Document,
+    Filter,
+    FindOptions,
+} from "mongodb";
+import {Socket} from "socket.io";
+
+import {
+    ClientToServerEvents,
+    InterServerEvents,
+    ServerToClientEvents,
+    SocketData,
+} from "../../../../../../../common/index.js";
+
+
+/**
+ * Custom socket type for Mongo Socket IO server.
+ */
+type MongoCustomSocket = Socket<
+    ClientToServerEvents,
+    ServerToClientEvents,
+    InterServerEvents,
+    SocketData
+>;
+
+/**
+ * Unique ID to represent each socket connection.
+ */
+type ConnectionId = string;
+
+
+/**
+ * Parameters for MongoDB queries.
+ */
+interface QueryParameters {
+    collectionName: string;
+    query: Filter<Document>;
+    options: FindOptions;
+}
+
+/**
+ * Options to connect to MongoDB database.
+ */
+interface DbOptions {
+    // Name of database.
+    database: string;
+    host: string;
+    port: number;
+}
+
+/**
+ * Timeout for emitting updates to the client.
+ */
+const CLIENT_UPDATE_TIMEOUT_MILLIS = 500;
+
+/**
+ * MongoDB change stream for a query, and a list of subscribed connections. Subscribed connections
+ * can include duplicates if the same connection subscribes to the same query multiple times.
+ */
+interface Watcher {
+    changeStream: ChangeStream;
+    subscribers: ConnectionId[];
+}
+
+export {
+    CLIENT_UPDATE_TIMEOUT_MILLIS,
+    ConnectionId,
+    DbOptions,
+    MongoCustomSocket,
+    QueryParameters,
+    Watcher,
+};
diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts
new file mode 100644
index 0000000000..159fb64759
--- /dev/null
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts
@@ -0,0 +1,80 @@
+import {
+    Document,
+    Filter,
+} from "mongodb";
+
+import {
+    QueryParameters,
+} from "./typings.js";
+
+
+/**
+ * Modifies query so that it can be used to filter change stream update events. Update events in
+ * MongoDB return the document in the "fullDocument" field, so function prepends "fullDocument" to
+ * each key in the query.
+ *
+ * Reference: https://www.mongodb.com/docs/manual/reference/change-events/update/#mongodb-data-update
+ *
+ * @param query
+ * @return Modified query.
+ */
+const convertQueryToChangeStreamFormat = (
+    query: Filter<Document>
+): Filter<Document> => {
+    const changeStreamQuery: Filter<Document> = {};
+    for (const key in query) {
+        if (Object.hasOwn(query, key)) {
+            changeStreamQuery[`fullDocument.${key}`] = query[key] as unknown;
+        }
+    }
+
+    return changeStreamQuery;
+};
+
+/**
+ * Generates a unique hash for a given query parameters.
+ *
+ * @param queryParams
+ * @return Unique hash for query parameters.
+ */
+const getQueryHash = (
+    queryParams: QueryParameters
+): string => {
+    return JSON.stringify(queryParams);
+};
+
+/**
+ * Recovers query parameters from the hash.
+ *
+ * @param queryHash
+ * @return
+ */
+const getQuery = (
+    queryHash: string
+): QueryParameters => {
+    const parsedValue: unknown = JSON.parse(queryHash);
+    return parsedValue as QueryParameters;
+};
+
+/**
+ * Removes the first instance of item from an array, if it exists.
+ *
+ * @param array
+ * @param item
+ */
+const removeItemFromArray = <T>(
+    array: T[],
+    item: T
+): void => {
+    const index = array.indexOf(item);
+    if (-1 !== index) {
+        array.splice(index, 1);
+    }
+};
+
+export {
+    convertQueryToChangeStreamFormat,
+    getQuery,
+    getQueryHash,
+    removeItemFromArray,
+};

From 8ed53de54cae5634bb5e67221dc1e61443949265 Mon Sep 17 00:00:00 2001
From: Marco <david.marcovitch@yscope.com>
Date: Mon, 23 Jun 2025 18:44:22 +0000
Subject: [PATCH 3/7] latest

---
 components/log-viewer-webui/server/src/app.ts     |  1 -
 .../MongoSocketIoServer/MongoWatcherCollection.ts |  4 ++--
 .../app/socket/MongoSocketIoServer/index.ts       | 15 ++++++---------
 3 files changed, 8 insertions(+), 12 deletions(-)

diff --git a/components/log-viewer-webui/server/src/app.ts b/components/log-viewer-webui/server/src/app.ts
index e1ee98b9f5..c523b34664 100644
--- a/components/log-viewer-webui/server/src/app.ts
+++ b/components/log-viewer-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 "./fastify-v2/plugins/app/socket/MongoSocketIoServer/index.js";
 import S3Manager from "./plugins/S3Manager.js";
 import exampleRoutes from "./routes/example.js";
 import queryRoutes from "./routes/query.js";
diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
index d76515ef82..77956f9cf3 100644
--- a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
@@ -30,10 +30,10 @@ class MongoWatcherCollection {
 
     /**
      * @param collectionName
-     * @param mongoDb
      * @param logger
+     * @param mongoDb
      */
-    constructor (collectionName: string, mongoDb: Db, logger: FastifyBaseLogger) {
+    constructor (collectionName: string, logger: FastifyBaseLogger, mongoDb: Db) {
         this.#collection = mongoDb.collection(collectionName);
         this.#logger = logger;
     }
diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
index 84791f8f7f..a2b3c18b84 100644
--- a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
@@ -60,21 +60,18 @@ 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 io
-     * @param mongoDb
      * @param logger
+     * @param mongoDb
      */
-    constructor (
+    private constructor (
         io: Server<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>,
-        mongoDb: Db,
-        logger: FastifyBaseLogger
+        logger: FastifyBaseLogger,
+        mongoDb: Db
     ) {
+        this.#io = io;
         this.#logger = logger;
         this.#mongoDb = mongoDb;
-        this.#io = io;
         this.#registerEventListeners();
     }
 
@@ -100,7 +97,7 @@ class MongoSocketIoServer {
             SocketData
         >(fastify.server);
 
-        return new MongoSocketIoServer(io, mongoDb, fastify.log);
+        return new MongoSocketIoServer(io, fastify.log, mongoDb);
     }
 
     /**

From c966314ea87e539f2d45605372b04ba2436588b9 Mon Sep 17 00:00:00 2001
From: Marco <david.marcovitch@yscope.com>
Date: Mon, 23 Jun 2025 18:58:25 +0000
Subject: [PATCH 4/7] latest

---
 .../fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
index a2b3c18b84..1148c45ef1 100644
--- a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
@@ -205,7 +205,7 @@ class MongoSocketIoServer {
         : MongoWatcherCollection {
         let watcherCollection = this.#collections.get(collectionName);
         if ("undefined" === typeof watcherCollection) {
-            watcherCollection = new MongoWatcherCollection(collectionName, this.#mongoDb, this.#logger);
+            watcherCollection = new MongoWatcherCollection(collectionName, this.#logger, this.#mongoDb);
             this.#logger.debug(`Initialize Mongo watcher collection:${collectionName}.`);
             this.#collections.set(collectionName, watcherCollection);
         }

From 7b82bd74e9e2c2406f3052b70b449aab98e1351a Mon Sep 17 00:00:00 2001
From: Marco <david.marcovitch@yscope.com>
Date: Mon, 23 Jun 2025 19:02:00 +0000
Subject: [PATCH 5/7] latest

---
 .../socket/MongoSocketIoServer/MongoWatcherCollection.ts | 2 +-
 .../plugins/app/socket/MongoSocketIoServer/index.ts      | 9 +++++++--
 .../plugins/app/socket/MongoSocketIoServer/utils.ts      | 4 +---
 3 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
index 77956f9cf3..d3f825323d 100644
--- a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
@@ -1,8 +1,8 @@
+import {FastifyBaseLogger} from "fastify";
 import type {
     Collection,
     Db,
 } from "mongodb";
-import {FastifyBaseLogger} from "fastify";
 
 import {QueryId} from "../../../../../../../common/index.js";
 import {
diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
index 1148c45ef1..7527dc2eee 100644
--- a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
@@ -4,6 +4,7 @@
 // Reference: https://github.com/socketio/socket.io/blob/main/examples/basic-crud-application/server/lib/todo-management/todo.handlers.ts
 
 import {
+    FastifyBaseLogger,
     FastifyInstance,
 } from "fastify";
 import fastifyPlugin from "fastify-plugin";
@@ -29,7 +30,6 @@ import {
     getQueryHash,
     removeItemFromArray,
 } from "./utils.js";
-import {FastifyBaseLogger} from "fastify";
 
 
 /**
@@ -79,6 +79,7 @@ class MongoSocketIoServer {
      * Creates a new MongoSocketIoServer.
      *
      * @param fastify
+     * @throws {Error} When MongoDB database not found
      * @return
      */
     static create (
@@ -205,7 +206,11 @@ class MongoSocketIoServer {
         : MongoWatcherCollection {
         let watcherCollection = this.#collections.get(collectionName);
         if ("undefined" === typeof watcherCollection) {
-            watcherCollection = new MongoWatcherCollection(collectionName, this.#logger, this.#mongoDb);
+            watcherCollection = new MongoWatcherCollection(
+                collectionName,
+                 this.#logger,
+                this.#mongoDb
+            );
             this.#logger.debug(`Initialize Mongo watcher collection:${collectionName}.`);
             this.#collections.set(collectionName, watcherCollection);
         }
diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts
index 159fb64759..6c17029dc6 100644
--- a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts
@@ -3,9 +3,7 @@ import {
     Filter,
 } from "mongodb";
 
-import {
-    QueryParameters,
-} from "./typings.js";
+import {QueryParameters} from "./typings.js";
 
 
 /**

From 639766b6f70b7ef2e3b5d44b73f015eca245e832 Mon Sep 17 00:00:00 2001
From: Marco <david.marcovitch@yscope.com>
Date: Mon, 23 Jun 2025 19:44:11 +0000
Subject: [PATCH 6/7] latest

---
 .../plugins/app/socket/MongoSocketIoServer/index.ts           | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
index 7527dc2eee..e2454af180 100644
--- a/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
+++ b/components/log-viewer-webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
@@ -79,8 +79,8 @@ class MongoSocketIoServer {
      * Creates a new MongoSocketIoServer.
      *
      * @param fastify
-     * @throws {Error} When MongoDB database not found
      * @return
+     * @throws {Error} When MongoDB database not found
      */
     static create (
         fastify: FastifyInstance
@@ -208,7 +208,7 @@ class MongoSocketIoServer {
         if ("undefined" === typeof watcherCollection) {
             watcherCollection = new MongoWatcherCollection(
                 collectionName,
-                 this.#logger,
+                this.#logger,
                 this.#mongoDb
             );
             this.#logger.debug(`Initialize Mongo watcher collection:${collectionName}.`);

From 76f1b4ccf7c9a3e9bf1ed716b59a85f0a4bfe7f9 Mon Sep 17 00:00:00 2001
From: Marco <david.marcovitch@yscope.com>
Date: Mon, 30 Jun 2025 16:56:31 +0000
Subject: [PATCH 7/7] latest

---
 .../MongoWatcherCollection.ts                 | 215 ++++++++++
 .../app/socket/MongoSocketIoServer/index.ts   | 389 ++++++++++++++++++
 .../app/socket/MongoSocketIoServer/typings.ts |  73 ++++
 .../app/socket/MongoSocketIoServer/utils.ts   |  78 ++++
 4 files changed, 755 insertions(+)
 create mode 100644 components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
 create mode 100644 components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
 create mode 100644 components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts
 create mode 100644 components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts

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
new file mode 100644
index 0000000000..d3f825323d
--- /dev/null
+++ b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts
@@ -0,0 +1,215 @@
+import {FastifyBaseLogger} from "fastify";
+import type {
+    Collection,
+    Db,
+} from "mongodb";
+
+import {QueryId} from "../../../../../../../common/index.js";
+import {
+    CLIENT_UPDATE_TIMEOUT_MILLIS,
+    MongoCustomSocket,
+    QueryParameters,
+    Watcher,
+} from "./typings.js";
+import {
+    convertQueryToChangeStreamFormat,
+    removeItemFromArray,
+} from "./utils.js";
+
+
+/**
+ * Provides watchers for MongoDB queries to a specific collection.
+ */
+class MongoWatcherCollection {
+    #collection: Collection;
+
+    #logger: FastifyBaseLogger;
+
+    // Active watchers
+    #queryIdtoWatcherMap: Map<QueryId, Watcher> = new Map();
+
+    /**
+     * @param collectionName
+     * @param logger
+     * @param mongoDb
+     */
+    constructor (collectionName: string, logger: FastifyBaseLogger, mongoDb: Db) {
+        this.#collection = mongoDb.collection(collectionName);
+        this.#logger = logger;
+    }
+
+    /**
+     * Checks if the collection is currently being referenced by any clients.
+     *
+     * @return True if the collection is referenced, false otherwise.
+     */
+    isReferenced (): boolean {
+        return 0 !== this.#queryIdtoWatcherMap.size;
+    }
+
+    /**
+     * Checks if a watcher exists for the given query ID.
+     *
+     * @param queryId
+     * @return True if a watcher exists, false otherwise.
+     */
+    hasWatcher (
+        queryId: QueryId,
+    ): boolean {
+        if ("undefined" === typeof this.#queryIdtoWatcherMap.get(queryId)) {
+            return false;
+        }
+
+        return true;
+    }
+
+    /**
+     * Unsubscribes a connection from a watcher. If the watcher has no more subscribers, it closes
+     * the change stream.
+     *
+     * @param queryId
+     * @param connectionId
+     * @return True if connection is last subcriber, false otherwise.
+     */
+    unsubscribe (queryId: number, connectionId: string): boolean {
+        const watcher = this.#queryIdtoWatcherMap.get(queryId);
+
+        if ("undefined" === typeof watcher) {
+            this.#logger.warn(`No watcher found for queryID:${queryId}`);
+
+            return false;
+        }
+
+        if (1 < watcher.subscribers.length) {
+            removeItemFromArray(watcher.subscribers, connectionId);
+
+            return false;
+        }
+
+        watcher.changeStream.close().catch((err: unknown) => {
+            this.#logger.error(err, `Error closing watcher for queryID:${queryId}`);
+        });
+        this.#queryIdtoWatcherMap.delete(queryId);
+
+        return true;
+    }
+
+    /**
+     * Adds connection to an existing watcher and joins the the room for the given query ID.
+     *
+     * @param queryId
+     * @param socket
+     */
+    async subscribe (
+        queryId: QueryId,
+        socket: MongoCustomSocket
+    ) {
+        const watcher = this.#queryIdtoWatcherMap.get(queryId);
+
+        if ("undefined" === typeof watcher) {
+            throw new Error(`No watcher found for queryId ${queryId}`);
+        }
+
+        watcher.subscribers.push(socket.id);
+        await socket.join(queryId.toString());
+    }
+
+    /**
+     * Creates a watcher for the given query.
+     *
+     * @param queryParams
+     * @param queryId
+     * @param emitUpdate
+     */
+    createWatcher (
+        queryParams: QueryParameters,
+        queryId: QueryId,
+        emitUpdate: (data: object[]) => void
+    ): void {
+        const watcherQuery = convertQueryToChangeStreamFormat(queryParams.query);
+        const mongoWatcher = this.#collection.watch(
+            [{$match: watcherQuery}],
+            {fullDocument: "updateLookup"}
+        );
+
+        const watcher: Watcher = {changeStream: mongoWatcher, subscribers: []};
+        this.#setupWatcherListener(watcher, queryParams, queryId, emitUpdate);
+        this.#queryIdtoWatcherMap.set(queryId, watcher);
+    }
+
+    /**
+     * Executes a query on the collection and retrieves matching documents.
+     *
+     * @param queryParameters
+     * @return
+     */
+    async find (
+        queryParameters: QueryParameters
+    ): Promise<object[]> {
+        const {query, options} = queryParameters;
+        try {
+            const documents = await this.#collection.find(query, options).toArray();
+            return documents;
+        } catch (error) {
+            this.#logger.error(error, "Error fetching data for query");
+
+            return [];
+        }
+    }
+
+    /**
+     * Sets up listener to emit updates to clients on change events.
+     *
+     * @param watcher
+     * @param queryParameters
+     * @param queryId
+     * @param emitUpdate
+     */
+    #setupWatcherListener (
+        watcher: Watcher,
+        queryParameters: QueryParameters,
+        queryId: QueryId,
+        emitUpdate: (data: object[]) => void
+    ) {
+        let lastEmitTime = 0;
+        let emitTimeout: NodeJS.Timeout | null = null;
+
+        const emitUpdateWithTimeout = async () => {
+            const currentTime = Date.now();
+            if (CLIENT_UPDATE_TIMEOUT_MILLIS <= currentTime - lastEmitTime) {
+                lastEmitTime = currentTime;
+                const data = await this.find(queryParameters);
+                emitUpdate(data);
+            } else if (null === emitTimeout) {
+                const delay = CLIENT_UPDATE_TIMEOUT_MILLIS - (currentTime - lastEmitTime);
+
+                emitTimeout = setTimeout(() => {
+                    emitTimeout = null;
+
+                    const fetchAndEmit = async () => {
+                        const data = await this.find(queryParameters);
+                        emitUpdate(data);
+                    };
+
+                    fetchAndEmit().catch((error: unknown) => {
+                        this.#logger.error(error, "Error in emitUpdatesWithTimeout");
+                    });
+                    lastEmitTime = Date.now();
+                }, delay);
+            }
+        };
+
+        watcher.changeStream.on("change", (change) => {
+            if ("invalidate" === change.operationType) {
+                this.#logger.info(`Change stream received invalidate event for queryID ${queryId}`);
+
+                return;
+            }
+            emitUpdateWithTimeout().catch((error: unknown) => {
+                this.#logger.error(error, "Error in emitUpdatesWithTimeout");
+            });
+        });
+    }
+}
+
+export default MongoWatcherCollection;
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
new file mode 100644
index 0000000000..e2454af180
--- /dev/null
+++ b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts
@@ -0,0 +1,389 @@
+/* eslint-disable max-lines */
+/* eslint-disable no-warning-comments */
+// TODO: Move listeners to a separate file to reduce lines
+// Reference: https://github.com/socketio/socket.io/blob/main/examples/basic-crud-application/server/lib/todo-management/todo.handlers.ts
+
+import {
+    FastifyBaseLogger,
+    FastifyInstance,
+} from "fastify";
+import fastifyPlugin from "fastify-plugin";
+import {Db} from "mongodb";
+import {Server} from "socket.io";
+
+import type {
+    ClientToServerEvents,
+    InterServerEvents,
+    QueryId,
+    Response,
+    ServerToClientEvents,
+    SocketData,
+} from "../../../../../../../common/index.js";
+import MongoWatcherCollection from "./MongoWatcherCollection.js";
+import {
+    ConnectionId,
+    MongoCustomSocket,
+    QueryParameters,
+} from "./typings.js";
+import {
+    getQuery,
+    getQueryHash,
+    removeItemFromArray,
+} from "./utils.js";
+
+
+/**
+ * Integrates MongoDB with Socket.IO to provide real-time updates for MongoDB queries.
+ *
+ * TODO: In current implementation, multiple queries in the same collection send updates over
+ * one socket with the same event name. A potential improvement would be to use different event
+ * names per query, limiting the number of events listeners triggered in the client.
+ */
+class MongoSocketIoServer {
+    #logger: FastifyBaseLogger;
+
+    #io: Server<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>;
+
+    // Collections with active queries.
+    #collections: Map<string, MongoWatcherCollection> = new Map();
+
+    // Mapping of active queries to their hashes.
+    #queryIdToQueryHashMap: Map<QueryId, string> = new Map();
+
+    // Mapping of connection IDs to the query IDs they are subscribed to. A connection can
+    // subscribe to the same queryID multiple times, so the list can contain duplicates.
+    #subscribedQueryIdsMap: Map<ConnectionId, QueryId[]> = new Map();
+
+    // Counter for generating unique query IDs.
+    #queryIdCounter: QueryId = 0;
+
+    readonly #mongoDb: Db;
+
+    /**
+     * @param io
+     * @param logger
+     * @param mongoDb
+     */
+    private constructor (
+        io: Server<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>,
+        logger: FastifyBaseLogger,
+        mongoDb: Db
+    ) {
+        this.#io = io;
+        this.#logger = logger;
+        this.#mongoDb = mongoDb;
+        this.#registerEventListeners();
+    }
+
+    /**
+     * Creates a new MongoSocketIoServer.
+     *
+     * @param fastify
+     * @return
+     * @throws {Error} When MongoDB database not found
+     */
+    static create (
+        fastify: FastifyInstance
+    ): MongoSocketIoServer {
+        const mongoDb = fastify.mongo.db;
+
+        if ("undefined" === typeof mongoDb) {
+            throw new Error("MongoDB database not found");
+        }
+
+        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.#logger.info(`New socket connected with ID:${socket.id}`);
+            socket.on("disconnect", this.#disconnectListener.bind(this, socket));
+            socket.on(
+                "collection::find::subscribe",
+                this.#collectionFindSubscribeListener.bind(this, socket)
+            );
+            socket.on(
+                "collection::find::unsubscribe",
+                this.#collectionFindUnsubscribeListener.bind(this, socket)
+            );
+        });
+    }
+
+    /**
+     * Listener for socket disconnection events.
+     *
+     * @param socket
+     */
+    async #disconnectListener (socket: MongoCustomSocket) {
+        this.#logger.info(`Socket:${socket.id} disconnected`);
+        const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socket.id);
+
+        if ("undefined" === typeof subscribedQueryIds) {
+            return;
+        }
+
+        for (const queryId of subscribedQueryIds) {
+            this.#unsubscribe(socket, queryId);
+        }
+
+        this.#subscribedQueryIdsMap.delete(socket.id);
+        this.#logger.debug(
+            "Subscribed queryIDs map" +
+            ` ${JSON.stringify(Array.from(this.#subscribedQueryIdsMap.entries()))}`
+        );
+    }
+
+    /**
+     * Checks if a collection exists in the MongoDB database.
+     *
+     * @param collectionName
+     * @return Whether the collection exists.
+     */
+    async #hasCollection (collectionName: string): Promise<boolean> {
+        const collections = await this.#mongoDb.listCollections().toArray();
+        return collections.some((collection) => collection.name === collectionName);
+    }
+
+    /**
+     * Adds the query ID to the connection's subscribed query IDs.
+     *
+     * @param queryId
+     * @param socketId
+     */
+    #addQueryIdToSubscribedList (queryId: QueryId, socketId: ConnectionId) {
+        const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socketId);
+        if ("undefined" === typeof subscribedQueryIds) {
+            this.#subscribedQueryIdsMap.set(socketId, [queryId]);
+
+            return;
+        }
+
+        this.#subscribedQueryIdsMap.set(socketId, [...subscribedQueryIds,
+            queryId]);
+    }
+
+    /**
+     * Gets query ID based on query parameters. If not found, creates a new ID.
+     *
+     * @param queryParams
+     * @return the query ID.
+     */
+    #getQueryId (queryParams: QueryParameters): number {
+        const queryHash = getQueryHash(queryParams);
+        for (const [queryId, hash] of this.#queryIdToQueryHashMap.entries()) {
+            if (hash === queryHash) {
+                return queryId;
+            }
+        }
+        const queryId = this.#queryIdCounter;
+        this.#queryIdToQueryHashMap.set(queryId, queryHash);
+
+        // JS is single threaded and ++ is atomic, so we can safely increment the global counter.
+        this.#queryIdCounter++;
+
+        return queryId;
+    }
+
+    /**
+     * Gets an existing watcher collection or creates a new one if it doesn't exist.
+     *
+     * @param collectionName
+     * @return The watcher collection instance.
+     */
+    #getOrCreateWatcherCollection (
+        collectionName: string
+    )
+        : MongoWatcherCollection {
+        let watcherCollection = this.#collections.get(collectionName);
+        if ("undefined" === typeof watcherCollection) {
+            watcherCollection = new MongoWatcherCollection(
+                collectionName,
+                this.#logger,
+                this.#mongoDb
+            );
+            this.#logger.debug(`Initialize Mongo watcher collection:${collectionName}.`);
+            this.#collections.set(collectionName, watcherCollection);
+        }
+
+        return watcherCollection;
+    }
+
+    /**
+     * Listener for subscribing to a find query. The client will receive updates whenever
+     * the query results change.
+     *
+     * @param socket
+     * @param requestArgs
+     * @param requestArgs.query
+     * @param requestArgs.options
+     * @param requestArgs.collectionName
+     * @param callback
+     */
+    async #collectionFindSubscribeListener (
+        socket: MongoCustomSocket,
+        requestArgs: {collectionName: string; query: object; options: object},
+        callback: (res: Response<{queryId: number; initialDocuments: object[]}>) => void
+    ): Promise<void> {
+        const {collectionName, query, options} = requestArgs;
+
+        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.#logger.error(`Collection ${collectionName} does not exist in MongoDB`);
+            callback({
+                error: `Collection ${collectionName} does not exist in MongoDB on server`,
+            });
+
+            return;
+        }
+
+        const watcherCollection = this.#getOrCreateWatcherCollection(collectionName);
+
+        const queryParameters: QueryParameters = {collectionName, query, options};
+        const queryId = this.#getQueryId(queryParameters);
+
+        await this.#subscribeToQuery(watcherCollection, queryParameters, queryId, socket);
+
+        const initialDocuments = await watcherCollection.find(queryParameters);
+        callback({data: {queryId, initialDocuments}});
+
+        this.#addQueryIdToSubscribedList(queryId, socket.id);
+        this.#logger.info(
+            `Socket:${socket.id} subscribed to query:${JSON.stringify(query)} ` +
+            `with options:${JSON.stringify(options)} ` +
+            `on collection:${collectionName} with ID:${queryId}`
+        );
+    }
+
+    /**
+     * Subscribes to query updates.
+     *
+     * @param watcherCollection
+     * @param queryParameters
+     * @param queryId
+     * @param socket
+     */
+    async #subscribeToQuery (
+        watcherCollection: MongoWatcherCollection,
+        queryParameters: QueryParameters,
+        queryId: QueryId,
+        socket: MongoCustomSocket
+    ): Promise<void> {
+        if (false === watcherCollection.hasWatcher(queryId)) {
+            const emitUpdate = (data: object[]) => {
+                this.#io.to(`${queryId}`).emit("collection::find::update", {queryId, data});
+            };
+
+            watcherCollection.createWatcher(queryParameters, queryId, emitUpdate);
+        }
+        await watcherCollection.subscribe(queryId, socket);
+    }
+
+    /**
+     * Unsubscribes from a query.
+     *
+     * @param socket
+     * @param queryId
+     */
+    #unsubscribe (socket: MongoCustomSocket, queryId: number) {
+        const queryHash: string | undefined = this.#queryIdToQueryHashMap.get(queryId);
+        if ("undefined" === typeof queryHash) {
+            this.#logger.error(`Query:${queryId} not found in query map`);
+
+            return;
+        }
+
+        const queryParams: QueryParameters = getQuery(queryHash);
+
+        const collection = this.#collections.get(queryParams.collectionName);
+        if ("undefined" === typeof collection) {
+            this.#logger.error(`${queryParams.collectionName} is missing from server`);
+
+            return;
+        }
+
+        const isLastSubscriber = collection.unsubscribe(queryId, socket.id);
+        this.#logger.info(`Socket:${socket.id} unsubscribed from query:${queryId}`);
+
+        if (isLastSubscriber) {
+            this.#logger.debug(`Query:${queryId} deleted from query map.`);
+            this.#queryIdToQueryHashMap.delete(queryId);
+        }
+
+        this.#logger.debug(
+            "Query ID to query hash map:" +
+            ` ${JSON.stringify(Array.from(this.#queryIdToQueryHashMap.entries()))}`
+        );
+
+        if (false === collection.isReferenced()) {
+            this.#logger.debug(`Collection:${queryParams.collectionName}` +
+            " deallocated from server.");
+            this.#collections.delete(queryParams.collectionName);
+        }
+    }
+
+    /**
+     * Listener for unsubscribing from a find query.
+     *
+     * @param socket
+     * @param requestArgs
+     * @param requestArgs.queryId
+     */
+    async #collectionFindUnsubscribeListener (
+        socket: MongoCustomSocket,
+        requestArgs: {queryId: number}
+    ): Promise<void> {
+        const {queryId} = requestArgs;
+        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.#logger.error(`Socket ${socket.id} is not subscribed to ${queryId}`);
+
+            return;
+        }
+
+        this.#unsubscribe(socket, queryId);
+        await socket.leave(queryId.toString());
+
+        removeItemFromArray(subscribedQueryIds, queryId);
+
+        this.#logger.debug(
+            `Subscribed queryIDs map ${
+                JSON.stringify(Array.from(this.#subscribedQueryIdsMap.entries()))}`
+        );
+    }
+}
+
+declare module "fastify" {
+    export interface FastifyInstance {
+        MongoSocketIoServer: MongoSocketIoServer;
+    }
+}
+
+export default fastifyPlugin(
+    (fastify) => {
+        fastify.decorate("MongoSocketIoServer", 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
new file mode 100644
index 0000000000..13d8635529
--- /dev/null
+++ b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts
@@ -0,0 +1,73 @@
+import {
+    ChangeStream,
+    Document,
+    Filter,
+    FindOptions,
+} from "mongodb";
+import {Socket} from "socket.io";
+
+import {
+    ClientToServerEvents,
+    InterServerEvents,
+    ServerToClientEvents,
+    SocketData,
+} from "../../../../../../../common/index.js";
+
+
+/**
+ * Custom socket type for Mongo Socket IO server.
+ */
+type MongoCustomSocket = Socket<
+    ClientToServerEvents,
+    ServerToClientEvents,
+    InterServerEvents,
+    SocketData
+>;
+
+/**
+ * Unique ID to represent each socket connection.
+ */
+type ConnectionId = string;
+
+
+/**
+ * Parameters for MongoDB queries.
+ */
+interface QueryParameters {
+    collectionName: string;
+    query: Filter<Document>;
+    options: FindOptions;
+}
+
+/**
+ * Options to connect to MongoDB database.
+ */
+interface DbOptions {
+    // Name of database.
+    database: string;
+    host: string;
+    port: number;
+}
+
+/**
+ * Timeout for emitting updates to the client.
+ */
+const CLIENT_UPDATE_TIMEOUT_MILLIS = 500;
+
+/**
+ * MongoDB change stream for a query, and a list of subscribed connections. Subscribed connections
+ * can include duplicates if the same connection subscribes to the same query multiple times.
+ */
+interface Watcher {
+    changeStream: ChangeStream;
+    subscribers: ConnectionId[];
+}
+
+export {
+    CLIENT_UPDATE_TIMEOUT_MILLIS,
+    ConnectionId,
+    DbOptions,
+    MongoCustomSocket,
+    QueryParameters,
+    Watcher,
+};
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
new file mode 100644
index 0000000000..6c17029dc6
--- /dev/null
+++ b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts
@@ -0,0 +1,78 @@
+import {
+    Document,
+    Filter,
+} from "mongodb";
+
+import {QueryParameters} from "./typings.js";
+
+
+/**
+ * Modifies query so that it can be used to filter change stream update events. Update events in
+ * MongoDB return the document in the "fullDocument" field, so function prepends "fullDocument" to
+ * each key in the query.
+ *
+ * Reference: https://www.mongodb.com/docs/manual/reference/change-events/update/#mongodb-data-update
+ *
+ * @param query
+ * @return Modified query.
+ */
+const convertQueryToChangeStreamFormat = (
+    query: Filter<Document>
+): Filter<Document> => {
+    const changeStreamQuery: Filter<Document> = {};
+    for (const key in query) {
+        if (Object.hasOwn(query, key)) {
+            changeStreamQuery[`fullDocument.${key}`] = query[key] as unknown;
+        }
+    }
+
+    return changeStreamQuery;
+};
+
+/**
+ * Generates a unique hash for a given query parameters.
+ *
+ * @param queryParams
+ * @return Unique hash for query parameters.
+ */
+const getQueryHash = (
+    queryParams: QueryParameters
+): string => {
+    return JSON.stringify(queryParams);
+};
+
+/**
+ * Recovers query parameters from the hash.
+ *
+ * @param queryHash
+ * @return
+ */
+const getQuery = (
+    queryHash: string
+): QueryParameters => {
+    const parsedValue: unknown = JSON.parse(queryHash);
+    return parsedValue as QueryParameters;
+};
+
+/**
+ * Removes the first instance of item from an array, if it exists.
+ *
+ * @param array
+ * @param item
+ */
+const removeItemFromArray = <T>(
+    array: T[],
+    item: T
+): void => {
+    const index = array.indexOf(item);
+    if (-1 !== index) {
+        array.splice(index, 1);
+    }
+};
+
+export {
+    convertQueryToChangeStreamFormat,
+    getQuery,
+    getQueryHash,
+    removeItemFromArray,
+};