diff --git a/components/webui/server/src/app.ts b/components/webui/server/src/app.ts index 1f4a35cafd..c523b34664 100644 --- 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 = 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/plugins/MongoSocketIoServer/MongoWatcherCollection.ts b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts similarity index 87% rename from components/webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts rename to components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/MongoWatcherCollection.ts index 6c2677f202..d3f825323d 100644 --- a/components/webui/server/src/plugins/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 = 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/plugins/MongoSocketIoServer/index.ts b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts similarity index 84% rename from components/webui/server/src/plugins/MongoSocketIoServer/index.ts rename to components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/index.ts index af14c0b760..cd79a52580 100644 --- a/components/webui/server/src/plugins/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; @@ -65,21 +63,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 fastify + * @param io + * @param logger * @param mongoDb */ - constructor (fastify: FastifyInstance, mongoDb: Db) { - this.#fastify = fastify; + private constructor ( + io: Server, + 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(); } @@ -87,14 +82,17 @@ class MongoSocketIoServer { * 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 { - 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,7 +122,14 @@ 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); } /** @@ -132,7 +137,7 @@ class MongoSocketIoServer { */ #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 { 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,7 +379,7 @@ class MongoSocketIoServer { requestArgs: {queryId: number} ): Promise { const {queryId} = requestArgs; - this.#fastify.log.debug( + this.#logger.debug( `Socket:${socket.id} requested unsubscription to query:${queryId}` ); @@ -378,7 +387,7 @@ class MongoSocketIoServer { 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 = 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/plugins/MongoSocketIoServer/typings.ts b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts similarity index 96% rename from components/webui/server/src/plugins/MongoSocketIoServer/typings.ts rename to components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/typings.ts index 2c1b1e0a96..13d8635529 100644 --- a/components/webui/server/src/plugins/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/plugins/MongoSocketIoServer/utils.ts b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts similarity index 72% rename from components/webui/server/src/plugins/MongoSocketIoServer/utils.ts rename to components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts index acdb8dd9ea..6c17029dc6 100644 --- a/components/webui/server/src/plugins/MongoSocketIoServer/utils.ts +++ b/components/webui/server/src/fastify-v2/plugins/app/socket/MongoSocketIoServer/utils.ts @@ -1,14 +1,9 @@ import { - Db, Document, Filter, - MongoClient, } from "mongodb"; -import { - DbOptions, - QueryParameters, -} from "./typings.js"; +import {QueryParameters} from "./typings.js"; /** @@ -59,27 +54,6 @@ const getQuery = ( 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 => { - 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. * @@ -100,6 +74,5 @@ export { convertQueryToChangeStreamFormat, getQuery, getQueryHash, - initializeMongoClient, removeItemFromArray, };