feat(new-webui): Mongo Framework Server - #814
Conversation
WalkthroughThis change introduces a new real-time, reactive MongoDB query subscription system to the log viewer web UI server. The update adds Socket.IO as a dependency and integrates a new Fastify plugin that allows clients to subscribe to MongoDB queries via WebSockets. The system manages query watchers, leverages MongoDB change streams, and emits updates to subscribed clients when query results change. Supporting utilities and TypeScript typings are included for query hashing, MongoDB connection, and event typing. The changes are modular, with new files for the main plugin, watcher management, utilities, and type definitions. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant SocketIoServer
participant MongoSocketIoServer
participant MongoWatcherCollection
participant MongoDB
Client->>SocketIoServer: Connect (WebSocket)
SocketIoServer->>MongoSocketIoServer: "collection::init" (collectionName)
MongoSocketIoServer->>MongoDB: Check collection existence
MongoDB-->>MongoSocketIoServer: Collection exists/does not exist
MongoSocketIoServer-->>SocketIoServer: Response (success/error)
Client->>SocketIoServer: "collection::find::toReactiveArray" (query, options)
SocketIoServer->>MongoSocketIoServer: Handle subscription
MongoSocketIoServer->>MongoWatcherCollection: getWatcher(query, queryId, socket)
MongoWatcherCollection->>MongoDB: Create or reuse ChangeStream
MongoWatcherCollection-->>MongoSocketIoServer: Watcher ready
MongoSocketIoServer-->>SocketIoServer: Response (queryId)
MongoWatcherCollection->>MongoDB: Listen for changes (ChangeStream)
MongoDB-->>MongoWatcherCollection: Change event
MongoWatcherCollection->>SocketIoServer: Emit "collection::find::update" (queryId, data) to subscribed clients
Client->>SocketIoServer: "collection::find::unsubscribe" (queryId)
SocketIoServer->>MongoSocketIoServer: Handle unsubscription
MongoSocketIoServer->>MongoWatcherCollection: unsubscribeFromWatcher(queryId, connectionId)
MongoWatcherCollection-->>MongoSocketIoServer: Unsubscribed/cleaned up
Client->>SocketIoServer: Disconnect
SocketIoServer->>MongoSocketIoServer: Handle disconnect
MongoSocketIoServer->>MongoWatcherCollection: Cleanup all subscriptions for client
✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (9)
components/log-viewer-webui/server/src/app.ts (1)
60-65: Consider conditional loading for test environments and ensure fail-safe error handling.Here you register the
MongoReplicaServerPluginunconditionally. If the test environment does not require a MongoDB connection or real-time communication, you could skip loading the plugin to speed up tests and avoid potential connection overhead. Also, make sure you handle any registration errors gracefully if the plugin’s underlying MongoDB connection fails.components/log-viewer-webui/server/src/plugins/MongoReplicaServerPlugin.ts (4)
50-62: Include host and port in the error message for better debugging.When a MongoDB connection error occurs, the thrown error message does not mention the host or port. Including them can ease troubleshooting in multi-environment setups.
- throw new Error("MongoDB connection error", {cause: e}); + throw new Error(`MongoDB connection error (host: ${host}, port: ${port})`, {cause: e});
64-81: Add access control in collection initialization if required.You rely on the user’s payload to identify the desired collection. If there is sensitive data, consider verifying user permissions before allowing clients to initialize certain collections.
83-96: Use consistent comparison style for improved clarity.Line 90 uses the exclamation operator (
!collection.isReferenced()). Per your coding guidelines, preferfalse == collection.isReferenced()for boolean checks in TypeScript files.- if (!collection.isReferenced()) { + if (false == collection.isReferenced()) {
98-116: Consider result size handling for large queries.As written,
.toArray()on large datasets could impact performance. You may benefit from pagination, limiting the returned documents, or streaming results. Let me know if you would like assistance implementing that.components/log-viewer-webui/server/src/plugins/MongoReplicaServerCollection.ts (4)
8-16: Improve documentation for getQueryHashThe function documentation contains a TODO comment and is missing proper parameter and return descriptions.
Update the JSDoc to better document the purpose, parameters, and return value:
/** - * // eslint-disable-next-line no-warning-comments - * TODO: Improve this? Think about security (other queries should not be able to kick others - * offline; maybe add a ref count then), performance, and collision chances. - * - * @param query - * @param options - * @return + * Generates a hash string from the query and options objects to uniquely identify a MongoDB query. + * Used for tracking change stream watchers associated with specific queries. + * + * @param query - The MongoDB query object + * @param options - The MongoDB query options object + * @return A string hash representing the combined query and options */
55-57: Add validation and type safety to the find methodThe
findmethod lacks input validation and type annotations for return value.-find (query: object, options: object) { - return this.collection.find(query, options); +/** + * Find documents in the collection that match the query + * + * @param query - The MongoDB query object + * @param options - The MongoDB query options object + * @return The MongoDB cursor for the query results + */ +find (query: object, options: object = {}) { + // Validate inputs + if (query === null || typeof query !== 'object') { + throw new Error('Query must be a valid object'); + } + return this.collection.find(query, options); }
62-62: Follow coding conventions for type checkingThe code uses string comparison with
"undefined"instead of directly checking if the variable is undefined.- if ("undefined" === typeof watcher) { + if (watcher === undefined) {
70-82: Consider adding a cleanup method for resource managementThe class has methods to add, remove, and check references, but lacks a comprehensive cleanup method to release all resources.
Add a cleanup method to close all watchers and reset the collection state:
+/** + * Closes all active watchers and resets the collection state + */ +async cleanup() { + // Close all watchers + const closePromises = Array.from(this.watchers.entries()).map(async ([queryHash, watcher]) => { + try { + await watcher.close(); + } catch (err) { + console.error(`Error closing watcher for queryHash ${queryHash}:`, err); + } + }); + + await Promise.all(closePromises); + + // Clear all watchers + this.watchers.clear(); + + // Reset count + this.count = 0; +}
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
components/log-viewer-webui/server/package-lock.jsonis excluded by!**/package-lock.json
📒 Files selected for processing (4)
components/log-viewer-webui/server/package.json(1 hunks)components/log-viewer-webui/server/src/app.ts(2 hunks)components/log-viewer-webui/server/src/plugins/MongoReplicaServerCollection.ts(1 hunks)components/log-viewer-webui/server/src/plugins/MongoReplicaServerPlugin.ts(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/app.tscomponents/log-viewer-webui/server/src/plugins/MongoReplicaServerPlugin.tscomponents/log-viewer-webui/server/src/plugins/MongoReplicaServerCollection.ts
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: lint-check (ubuntu-latest)
- GitHub Check: lint-check (macos-latest)
🔇 Additional comments (6)
components/log-viewer-webui/server/package.json (1)
33-33: Confirm compatibility of newly added dependency.The addition of
"socket.io": "^4.8.1"is consistent with introducing WebSocket-based communication. However, if your environment or other dependencies have special version requirements, please verify that this version does not introduce any compatibility issues or vulnerabilities.components/log-viewer-webui/server/src/app.ts (1)
11-11: Good integration of the MongoReplicaServerPlugin.Importing the plugin here seems appropriate. No immediate issues are observed.
components/log-viewer-webui/server/src/plugins/MongoReplicaServerPlugin.ts (1)
198-220: Well-structured plugin decoration.Decorating Fastify with the
MongoReplicaServerinstance is straightforward and keeps your code modular. No immediate issues are found here.components/log-viewer-webui/server/src/plugins/MongoReplicaServerCollection.ts (3)
26-30: LGTM: Constructor initializes properties appropriatelyThe constructor correctly initializes the count, collection, and watchers map.
51-53: LGTM: Boolean check follows coding guidelinesThe
isReferencedmethod uses0 < this.countwhich aligns with the coding guideline to preferfalse == <expression>rather than!<expression>.
73-81: LGTM: Good error handling in removeWatcher methodThe method properly handles both the case when a watcher exists and when it doesn't, including error handling for the close operation.
| * @param options | ||
| * @return | ||
| */ | ||
| const getQueryHash = (query: object, options: object): string => JSON.stringify({query, options}); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Improve hash generation to prevent collisions
The current implementation of getQueryHash uses JSON.stringify which doesn't guarantee consistent ordering of object keys. This could lead to different hashes for semantically identical queries with properties in different orders.
Consider using a more robust hashing approach that:
- Ensures consistent ordering of keys
- Handles potential circular references
-const getQueryHash = (query: object, options: object): string => JSON.stringify({query, options});
+const getQueryHash = (query: object, options: object): string => {
+ // Sort keys for consistent hash generation
+ const sortObjectKeys = (obj: Record<string, any>): Record<string, any> => {
+ return Object.keys(obj).sort().reduce((result, key) => {
+ const value = obj[key];
+ result[key] = value && typeof value === 'object' && !Array.isArray(value)
+ ? sortObjectKeys(value)
+ : value;
+ return result;
+ }, {} as Record<string, any>);
+ };
+
+ return JSON.stringify({
+ query: sortObjectKeys(query as Record<string, any>),
+ options: sortObjectKeys(options as Record<string, any>)
+ });
+};📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const getQueryHash = (query: object, options: object): string => JSON.stringify({query, options}); | |
| const getQueryHash = (query: object, options: object): string => { | |
| // Sort keys for consistent hash generation | |
| const sortObjectKeys = (obj: Record<string, any>): Record<string, any> => { | |
| return Object.keys(obj).sort().reduce((result, key) => { | |
| const value = obj[key]; | |
| result[key] = value && typeof value === 'object' && !Array.isArray(value) | |
| ? sortObjectKeys(value) | |
| : value; | |
| return result; | |
| }, {} as Record<string, any>); | |
| }; | |
| return JSON.stringify({ | |
| query: sortObjectKeys(query as Record<string, any>), | |
| options: sortObjectKeys(options as Record<string, any>) | |
| }); | |
| }; |
…separate folder and rename as MongoReplicaServer; fixed the async create() not being `await`ed.
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (8)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (6)
28-33: Ensure potential environment variable support for MongoDB connection details.While hard-coding or passing host/port directly works, you may want to consider environment variables or a configuration file for more flexibility (especially in production). This would reduce the need to recompile or redeploy for different environments.
35-49: Add fallback or validation for port string conversions.In case the provided port is not a valid string or does not parse correctly, consider adding validation or a fallback to avoid runtime errors.
51-63: Consider improving MongoClient error handling or retries.Currently, if the client connection fails even temporarily, the error is thrown, stopping the server from functioning. Adding retry logic or graceful fallback could improve reliability.
65-82: Validate collectionName payload to enhance security.Clients can supply arbitrary strings under
collectionName, which might lead to unintended or potentially malicious queries. Consider validating or sanitizing the collection name before usage.Would you like help drafting a validation approach (e.g., whitelisting, regex-based checks) for collection names?
91-91: Adhere to coding guideline “Prefer false == over !”.You are using a “not” operator (
!collection.isReferenced()). Per the guidelines, it should befalse == collection.isReferenced().- if (!collection.isReferenced()) { + if (false == collection.isReferenced()) {
166-196: Configure Socket.IO for production security (e.g., CORS).Using a default socket server might expose your application to cross-domain requests. Consider specifying CORS or authorization checks, especially if the server is publicly accessible.
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/MongoReplicaServerCollection.ts (2)
17-18: Strengthen query-hash collision avoidance.Relying on a simple JSON stringification for the query hash might lead to collisions if two distinct queries produce equivalent JSON strings. If collisions are rare but possible, consider an additional salt or hashing algorithm like SHA-1 for precision.
59-68: Validate watchers for large or complex queries.Change streams can be expensive if the queries are very large or frequent. Consider limiting watchers to ensure the server is not overwhelmed.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
components/log-viewer-webui/server/package-lock.jsonis excluded by!**/package-lock.json
📒 Files selected for processing (3)
components/log-viewer-webui/server/src/app.ts(2 hunks)components/log-viewer-webui/server/src/plugins/MongoReplicaServer/MongoReplicaServerCollection.ts(1 hunks)components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- components/log-viewer-webui/server/src/app.ts
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.tscomponents/log-viewer-webui/server/src/plugins/MongoReplicaServer/MongoReplicaServerCollection.ts
⏰ Context from checks skipped due to timeout of 90000ms (3)
- GitHub Check: build (macos-latest)
- GitHub Check: lint-check (ubuntu-latest)
- GitHub Check: lint-check (macos-latest)
🔇 Additional comments (3)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (1)
208-221: Plugin initialization looks good.The overall structure for registering the MongoReplicaServer with Fastify aligns well with plugin best practices.
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/MongoReplicaServerCollection.ts (2)
26-30: Constructor sets up watchers and reference counter cleanly.No immediate concerns; the structure is straightforward and aligns with intended design.
70-82:Details
❓ Verification inconclusive
Confirm watcher closure on final collection removal.
If the containing collection is garbage collected, watchers might remain open. Before removing the collection from memory, consider closing all watchers to avoid memory leaks or open streams.
🏁 Script executed:
#!/bin/bash # Explore references to `this.collections.delete(collectionName)` and confirm watchers are closed. rg -A 5 $'this.collections.delete(collectionName)'Length of output: 51
Subject: Verify that watchers are explicitly closed when a collection is finally removed
The current search for references to
this.collections.delete(collectionName)produced no output, so we cannot confirm automatically that watchers are closed during final collection deletion. Please manually verify that all watchers linked to a collection are fully closed before the collection is garbage collected. If needed, consider adding explicit code to close any open watchers prior to or during the collection removal process.
- File: components/log-viewer-webui/server/src/plugins/MongoReplicaServer/MongoReplicaServerCollection.ts (Lines 70-82)
- Action: Manually inspect the collection deletion flow to ensure that
removeWatcheris invoked as needed, and check for any cases where watchers might remain open after the collection is removed.
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (2)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (2)
84-90: [Duplicate from previous review] Close watchers properly on disconnect if no queries remain.Close watchers upon disconnect if no queries remain. Although the disconnect listener decrements the reference count, it does not explicitly close watchers. If the user never unsubscribes, watchers may remain active until
isReferenced()returns false. Consider automatically clearing watchers for that socket if the reference count drops to zero.Also applies to: 92-97
143-144: [Duplicate from previous review] Debounce the "change" handler.Repeated change events for large or rapidly updating collections can degrade performance. The inline comment references a “FIXME” regarding debouncing. Consider implementing a debounce strategy to avoid overwhelming downstream consumers with frequent updates.
🧹 Nitpick comments (4)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (4)
70-79: Use a more conventional undefined check.In TypeScript, it is more typical to check
if (collection === undefined)or simplyif (!collection). Using the string-literal'undefined' === typeof collectionis unusual and may reduce readability. Consider switching to a standard approach.
91-91: Adhere to coding guidelines: preferfalse == expressionrather than!expression.The current condition uses the logical NOT operator. According to the project's guidelines, consider using
false == <expression>for consistency:- if (!collection.isReferenced()) { + if (false == collection.isReferenced()) {
170-199: Optional improvement for event registration logic.Enumerating each event-listener pair in an array is effective but can become unwieldy as the application grows. Consider extracting this mapping into a typed configuration object or using a public method to attach these listeners, improving maintainability and making it easier to expand or modify event handling in future updates.
203-225: Consider adding automated tests for plug-in functionality.This plugin provides complex real-time interactions with MongoDB. A dedicated test suite would help ensure stability and spot regressions early. Tests for events, watcher handling, and error scenarios (e.g., failed connections) are recommended for robust coverage.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts
⏰ Context from checks skipped due to timeout of 90000ms (3)
- GitHub Check: lint-check (macos-latest)
- GitHub Check: lint-check (ubuntu-latest)
- GitHub Check: build (macos-latest)
|
I am still looking through but just for linting to start
For some of the linting failures, It looks like they are related to missing types for socket io. They have some docs here that may help for adding missing types - https://socket.io/docs/v4/typescript/ |
There was a problem hiding this comment.
Actionable comments posted: 9
♻️ Duplicate comments (2)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (2)
126-139: 🛠️ Refactor suggestionClose watchers upon disconnect if no queries remain.
The disconnect listener decrements the reference count but does not explicitly close watchers. If the user never unsubscribes from queries, watchers may remain open until
isReferenced()returns false. Consider automatically clearing all watchers for that socket if the ref count drops to zero.#getCollectionDisconnectListener (socket: Socket): EventListener<"disconnect"> { return () => { this.#fastify.log.info(`Socket disconnected: ${socket.id}`); const {collectionName} = socket.data as {collectionName: string}; const collection = this.#collections.get(collectionName); if ("undefined" !== typeof collection) { + // Clean up any remaining watchers for this socket + collection.removeAllWatchersForSocket(socket.id); collection.refRemove(); if (!collection.isReferenced()) { this.#fastify.log.info(`Collection ${collectionName} removed`); this.#collections.delete(collectionName); } } }; }Note: This requires adding a new method
removeAllWatchersForSocketto theMongoReplicaServerCollectionclass.
194-200: 🛠️ Refactor suggestionConsider debouncing the "change" handler.
The inline comment notes the need for debouncing. Repeated "change" events for large or rapidly updating collections could degrade performance.
Implement debouncing to prevent excessive updates:
- // eslint-disable-next-line @typescript-eslint/no-misused-promises - watcher.on("change", async () => { - // eslint-disable-next-line no-warning-comments - // FIXME: this should be debounced - socket.emit("collection::find::update", { - data: await collection.find(query, options).toArray(), - }); - }); + // Create debounced update function + let updateTimeout: NodeJS.Timeout | null = null; + const debouncedUpdate = async () => { + if (updateTimeout) { + clearTimeout(updateTimeout); + } + updateTimeout = setTimeout(async () => { + try { + const data = await collection.find(query, options).toArray(); + socket.emit("collection::find::update", { data }); + } catch (error) { + this.#fastify.log.error(`Error in change update: ${error}`); + socket.emit("error", { message: "Failed to update data" }); + } + }, 100); // 100ms debounce time + }; + + // eslint-disable-next-line @typescript-eslint/no-misused-promises + watcher.on("change", debouncedUpdate);
🧹 Nitpick comments (2)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (2)
1-47: Add class-level JSDoc documentation to improve maintainability and readability.The imports and interface definitions are well-structured, but the file lacks a comprehensive class description at the top level. Adding JSDoc documentation would help future developers understand the purpose and architecture of this server implementation.
+/** + * MongoReplicaServer plugin for Fastify + * + * This module implements a Socket.IO based server that provides real-time + * data synchronization with MongoDB collections. It allows clients to + * subscribe to collection changes and receive updates when documents change. + * + * The plugin is part of a new framework built on MongoCDC to replace + * the existing usage of Meteor's publish-subscribe framework. + */ import {FastifyInstance} from "fastify"; import fastifyPlugin from "fastify-plugin";
49-62: Add JSDoc comments to explain the class purpose and constructor.This class lacks documentation to explain its purpose and how it should be used. Adding JSDoc comments would improve maintainability.
+/** + * Manages connections to MongoDB replica sets and handles real-time data + * synchronization through Socket.IO. + * + * This class maintains a collection of MongoDB collections and provides + * methods for querying and subscribing to changes in these collections. + */ class MongoReplicaServer { #fastify: FastifyInstance; #collections: Map<string, MongoReplicaServerCollection>; #mongoDb: Db; + /** + * Creates a new MongoReplicaServer instance. + * + * @param fastify - The Fastify instance to attach to + * @param mongoDb - The MongoDB database instance + */ constructor ({fastify, mongoDb}: {fastify: FastifyInstance; mongoDb: Db}) { this.#fastify = fastify; this.#collections = new Map(); this.#mongoDb = mongoDb; this.#initializeSocketServer(fastify.server); }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: lint-check (ubuntu-latest)
- GitHub Check: lint-check (macos-latest)
junhaoliao
left a comment
junhaoliao
left a comment
There was a problem hiding this comment.
as discussed offline, let's remove all references to the name "replica". e.g., MongoReplicaServer -> FastifyMongoServer (i added "Fastify" in my proposal here because the name "MongoServer" alone can be ambiguous)
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (8)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (8)
100-112:⚠️ Potential issueAdd cleanup when MongoDB connection fails.
The error handling doesn't properly clean up resources if the connection fails. The MongoClient should be closed in the catch block to prevent resource leaks.
static async initializeMongoClient ( {database, host, port}: {database: string; host: string; port: string} ): Promise<Db> { const mongoUri = `mongodb://${host}:${port}`; const mongoClient = new MongoClient(mongoUri); try { await mongoClient.connect(); return mongoClient.db(database); } catch (e) { + await mongoClient.close().catch(closeError => { + console.error("Failed to close MongoDB connection after error", closeError); + }); throw new Error("MongoDB connection error", {cause: e}); } }
137-150: 🛠️ Refactor suggestionClose watchers upon disconnect if no queries remain.
The disconnect listener decrements the reference count but does not explicitly close watchers. If the user never unsubscribes from queries, watchers may remain open until
isReferenced()returns false.#getCollectionDisconnectListener (socket: CustomSocket) { return () => { this.#fastify.log.info(`Socket disconnected: ${socket.id}`); const {collectionName} = socket.data as {collectionName: string}; const collection = this.#collections.get(collectionName); if ("undefined" !== typeof collection) { + // Clean up any remaining watchers for this socket + collection.clearWatchersForSocket(socket.id); collection.refRemove(); if (!collection.isReferenced()) { this.#fastify.log.info(`Collection ${collectionName} removed`); this.#collections.delete(collectionName); } } }; }Note: This requires implementing a
clearWatchersForSocketmethod in theMongoReplicaServerCollectionclass to track watchers by socket ID.
187-194: 🛠️ Refactor suggestionImplement debouncing for the "change" handler.
The inline comment indicates the need for debouncing. Without it, repeated "change" events for large or rapidly updating collections will degrade performance and potentially flood clients with updates.
- // eslint-disable-next-line @typescript-eslint/no-misused-promises - watcher.on("change", async () => { - // eslint-disable-next-line no-warning-comments - // FIXME: this should be debounced - socket.emit("collection::find::update", { - data: await collection.find(query, options).toArray(), - }); - }); + // Implement debounced change handler + let debounceTimer: NodeJS.Timeout | null = null; + // eslint-disable-next-line @typescript-eslint/no-misused-promises + watcher.on("change", async () => { + if (debounceTimer) { + clearTimeout(debounceTimer); + } + + debounceTimer = setTimeout(async () => { + try { + const data = await collection.find(query, options).toArray(); + socket.emit("collection::find::update", { data }); + } catch (error) { + this.#fastify.log.error(`Error fetching updated data: ${error}`); + } + debounceTimer = null; + }, 300); // 300ms debounce time - adjust as needed + });
195-198: 🛠️ Refactor suggestionAdd error handling for initial data fetch.
The initial data fetch lacks error handling, which could lead to unhandled exceptions if the query fails.
- socket.emit("collection::find::update", { - data: await collection.find(query, options).toArray(), - }); + try { + const initialData = await collection.find(query, options).toArray(); + socket.emit("collection::find::update", { + data: initialData, + }); + } catch (error) { + this.#fastify.log.error(`Error in initial data fetch: ${error}`); + socket.emit("error", { message: "Failed to fetch initial data" }); + }
202-215: 🛠️ Refactor suggestionImprove unsubscribe handling with validation and feedback.
The unsubscribe handler lacks validation and client feedback, which could make debugging difficult.
#getCollectionFindUnsubscribeListener (socket: CustomSocket) : ClientToServerEvents["collection::find::unsubscribe"] { return ({queryHash}) => { + // Validate queryHash + if (false == queryHash || typeof queryHash !== 'string') { + this.#fastify.log.error(`Invalid queryHash: ${queryHash}`); + socket.emit('error', { message: 'Invalid queryHash' }); + return; + } + const {collectionName} = socket.data as {collectionName: string}; this.#fastify.log.info(`Collection name ${collectionName} requested unsubscription`); const collection = this.#collections.get(collectionName); if ("undefined" === typeof collection) { + socket.emit('error', { message: 'Collection not initialized' }); return; } - collection.removeWatcher(queryHash); + try { + const removed = collection.removeWatcher(queryHash); + if (removed) { + socket.emit('collection::find::unsubscribed', { queryHash }); + } else { + socket.emit('error', { message: 'Watcher not found' }); + } + } catch (error) { + this.#fastify.log.error(`Error removing watcher: ${error}`); + socket.emit('error', { message: 'Failed to unsubscribe' }); + } }; }
227-237: 🛠️ Refactor suggestionAdd error handling and cleanup for the plugin.
The plugin lacks error handling for server creation and cleanup logic for when the plugin is unregistered, which could lead to resource leaks.
const MongoReplicaServerPlugin = async ( app: FastifyInstance, options: {host: string; port: number; database: string} ) => { - await MongoReplicaServer.create({ - fastify: app, - host: options.host, - port: options.port.toString(), - database: options.database, - }); + let server; + try { + server = await MongoReplicaServer.create({ + fastify: app, + host: options.host, + port: options.port.toString(), + database: options.database, + }); + + // Add cleanup logic when Fastify closes + app.addHook('onClose', async () => { + app.log.info('Closing MongoDB connections...'); + // Add method to MongoReplicaServer to close MongoDB connections + // await server.close(); + }); + } catch (error) { + app.log.error(`Failed to create MongoDB replica server: ${error}`); + throw error; + } };
84-98: 🛠️ Refactor suggestionCorrect type inconsistency for port parameter.
The port parameter is defined as a string in the
createmethod, but as a number in the plugin function (line 229). This type inconsistency could lead to type errors.static async create ({ fastify, database, host, port, }: { fastify: FastifyInstance; database: string; host: string; - port: string; + port: number; }): Promise<MongoReplicaServer> { - const mongoDb = await MongoReplicaServer.initializeMongoClient({database, host, port}); + const mongoDb = await MongoReplicaServer.initializeMongoClient({database, host, port: port.toString()}); return new MongoReplicaServer({fastify, mongoDb}); }
152-168: 🛠️ Refactor suggestionAdd validation for collection name.
The collection initialization logic lacks validation for the collection name, which could lead to security issues or unexpected behavior.
#getCollectionInitListener (socket: CustomSocket): ClientToServerEvents["collection::init"] { return ({collectionName}) => { + // Validate collection name + if (false == collectionName || typeof collectionName !== 'string' || !collectionName.trim()) { + this.#fastify.log.error(`Invalid collection name requested: ${collectionName}`); + socket.emit('error', { message: 'Invalid collection name' }); + return; + } + this.#fastify.log.info(`Collection name ${collectionName} requested`); let collection = this.#collections.get(collectionName); if ("undefined" === typeof collection) { collection = new MongoReplicaServerCollection( this.#mongoDb, collectionName ); this.#collections.set(collectionName, collection); } collection.refAdd(); socket.data.collectionName = collectionName; }; }
🧹 Nitpick comments (2)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (2)
1-69: Consider adding ServerError type for consistent error handling.The error handling across the codebase is inconsistent. Creating a standardized error type would improve code consistency and client error handling.
+/** + * Standardized server error type + */ +interface ServerError { + message: string; + code?: string; + details?: unknown; +} + type ClientToServerEvents = { "disconnect": (reqArgs: never) => void; "collection::init": (reqArgs: { collectionName: string; }) => void; "collection::find::toArray": ( reqArgs: { query: object; options: object; }, callback: (respArgs: { data: Document[]; } | { - error: string; + error: ServerError; }) => void ) => Promise<void>; // Update other error types similarly...This would allow for more detailed error information to be passed to clients, including error codes, and would make error handling more consistent across the codebase.
218-226: Improve JSDoc comments for the plugin.The JSDoc comments for MongoReplicaServerPlugin are minimal and could be improved with more detailed descriptions of parameters and the plugin's purpose.
/** - * MongoDB replica set plugin for Fastify. + * MongoDB replica set plugin for Fastify. This plugin establishes a connection to a MongoDB + * replica set and sets up Socket.IO handlers for real-time collection queries and subscriptions. + * It replaces the existing Meteor publish-subscribe framework with a more efficient solution + * based on MongoDB Change Data Capture. * - * - * @param app - * @param options - * @param options.database - * @param options.host - * @param options.port + * @param app - The Fastify instance to register the plugin with + * @param options - Configuration options for the MongoDB connection + * @param options.database - The name of the MongoDB database to connect to + * @param options.host - The hostname of the MongoDB server + * @param options.port - The port number of the MongoDB server + * @returns {Promise<void>} - A promise that resolves when the plugin is registered */
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: lint-check (ubuntu-latest)
- GitHub Check: lint-check (macos-latest)
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (1)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (1)
146-167: Missing "collection::find::toArray" event registration.The client-to-server events define a
"collection::find::toArray"event, but it is never registered in the#initializeSocketServermethod. This prevents clients from calling this functionality.
🧹 Nitpick comments (3)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts (1)
183-183: Use 'false == condition' instead of '!condition' to comply with coding guidelines.Replace:
if (!collection.isReferenced()) { ... }with:
-if (!collection.isReferenced()) { +if (false == collection.isReferenced()) {components/log-viewer-webui/server/src/plugins/MongoReplicaServer/MongoReplicaServerCollection.ts (2)
8-11: Address the security TODO comment.The in-file TODO hints at potential security/enforcement improvements (e.g. ensuring watchers cannot be undesirably terminated by other clients). Consider access controls or user-bound watchers to mitigate abuse and collisions.
Would you like help drafting a secure design that integrates user-specific watchers, preventing unauthorized unsubscriptions?
130-137: Use the Fastify logger for consistent logging.The code uses
console.errorandconsole.warnfor watcher closure and missing watcher warnings. For consistency and improved observability, replace them withthis.#fastify.log.errororthis.#fastify.log.warn.- console.error(`Error closing watcher for queryHash ${queryHash}:`, err); + this.#fastify.log.error(`Error closing watcher for queryHash ${queryHash}: ${err}`); - console.warn(`No watcher found for queryHash ${queryHash}`); + this.#fastify.log.warn(`No watcher found for queryHash ${queryHash}`);
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/MongoReplicaServerCollection.ts(1 hunks)components/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/MongoReplicaServer/MongoReplicaServerCollection.tscomponents/log-viewer-webui/server/src/plugins/MongoReplicaServer/index.ts
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: lint-check (ubuntu-latest)
- GitHub Check: lint-check (macos-latest)
There was a problem hiding this comment.
I did my first pass of the code.
Two larger comments.
- It looks like each client can only subscribe to a single collection at once. Is that intended? It may make sense to add support to subscribe to multiple collections. Let me know
- Right now it looks like the client needs to disconnect to unsubscribe from a collection. Maybe it makes senses to add a "collection::unsubscribe" command, so the client can remove the collection itself?.
I will more at style in the next review.
Edit per discussion - The client is set up to open multiple sockets. As a result, it can support multiple collections. We can look into moving to a single socket at a later date.
There was a problem hiding this comment.
We should probably register await fastify.register(fastifySocketIO);, and use their existing plugin, instead of setting up socket.io with fastify.server
There was a problem hiding this comment.
For now we will do our own registration, since https://www.npmjs.com/package/fastify-socket.io does not support Fastify v5 or above.
|
Here is an option for the client side using only one socket, but maintaining something close to the current interface. Not neccesary for now, but potential later optimization. (1a) shared socket We could have a new function on the client then when creating new collection on the client, it uses the shared socket across all components Then for the cursor for the component could listen on something like this where the hash is the collection, query, options. 1b) Even if we dont do shared socket, it may still sense to still use a different event name for reactive arrays from the same client from the same collection. Something like this.socket.on("collection::find::update${hash})", where the hash is the query, option. It looks like if the client has multiple queries to the same collection, the notification will collide on the same event since they have the same name? Another option (1c) is maybe to force only one cursor per collection (2) Rooms. |
(1a) we should do shared socket on the client side given it's been fairly clear how it can be achived |
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (7)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.ts (1)
20-21: URI should include the database for correct auth / read preferences
const mongoUri = \mongodb://${host}:${port}`;omits the database name, which forces the driver to issue a secondisMastercall when selectingdb(database). Including/${database}` in the URI simplifies the handshake and respects per‑DB authentication rules.components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts (3)
238-243: Style: follow project guideline “false == ” instead of negationCoding‑guidelines specify avoiding the
!operator. Replace:if (!exists) {with:
if (false == exists) {This pops up in several places (
!exists,!collection, etc.). A quick search‑replace will keep lint happy.
284-300: Race‑prone manual key generation
#getQueryIdwalksMapentries and computesMath.max(...) + 1. Under concurrent awaits two subscribers could grab the same id. An auto‑incrementing field is simpler and atomic:private nextQueryId = 0; private #getQueryId(...) { for (…) return existing; return this.nextQueryId++; }
178-199: Sequentialawaitinside for‑loops slows disconnection
#collectionDisconnectListenerawaits inside nestedforloops. If a client was subscribed to many queries the disconnect stalls. Gather thePromises andawait Promise.all(...)instead to clean up in parallel.components/log-viewer-webui/server/src/plugins/FastifyMongoServer/MongoServerCollection.ts (3)
108-116: Duplicate subscriber IDs possible – favour aSet
watcher.subscribers.push(connectionId)can insert the same id multiple times if a client re‑subscribes. Replace the array withSet<string>to guarantee uniqueness and O(1) removals.
140-142: Prefer Fastify logger overconsole.errorDirect
consolecalls bypass Fastify’s logging levels/serialisers. Inject a logger or pass it in the constructor so errors integrate with the server’s log pipeline.
69-74: Reference counter underflow guard is silentA warning is printed but the counter can still go negative if
refRemove()is mis‑used multiple times. Consider throwing or at least keepingthis.count = 0to avoid negative numbers that later breakisReferenced().
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
components/log-viewer-webui/server/src/app.ts(2 hunks)components/log-viewer-webui/server/src/plugins/FastifyMongoServer/MongoServerCollection.ts(1 hunks)components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts(1 hunks)components/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.ts(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- components/log-viewer-webui/server/src/app.ts
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.tscomponents/log-viewer-webui/server/src/plugins/FastifyMongoServer/MongoServerCollection.tscomponents/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts
🧬 Code Graph Analysis (1)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts (1)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.ts (1)
initializeMongoClient(31-31)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: lint-check (ubuntu-latest)
- GitHub Check: lint-check (macos-latest)
| const initializeMongoClient = async function ( | ||
| database: string, | ||
| host: string, | ||
| port: string | ||
| ): Promise<Db> { | ||
| const mongoUri = `mongodb://${host}:${port}`; | ||
| const mongoClient = new MongoClient(mongoUri); | ||
| try { | ||
| await mongoClient.connect(); | ||
|
|
||
| return mongoClient.db(database); | ||
| } catch (e) { | ||
| throw new Error("MongoDB connection error", {cause: e}); | ||
| } | ||
| }; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Consider returning a cleanup handle to avoid leaked Mongo connections
initializeMongoClient opens a long‑lived MongoClient but never exposes a way for callers (or Fastify’s onClose) to close it. In production this can leave idle sockets hanging around after hot‑reloads or test runs, eventually exhausting the MongoDB connection pool.
-const initializeMongoClient = async function (
+const initializeMongoClient = async function (
database: string,
host: string,
port: string
-): Promise<Db> {
+): Promise<{db: Db; close: () => Promise<void>}> {
…
- await mongoClient.connect();
- return mongoClient.db(database);
+ await mongoClient.connect();
+ return {
+ db: mongoClient.db(database),
+ close: () => mongoClient.close(),
+ };
}Fastify plugins can then register:
fastify.addHook("onClose", async () => {
await mongoHandle.close();
});| #initializeSocketServer (httpServer: HttpServer) { | ||
| const io = new Server< | ||
| ClientToServerEvents, | ||
| ServerToClientEvents, | ||
| InterServerEvents, | ||
| SocketData | ||
| >(httpServer); | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Socket.IO initialisation lacks CORS and path hardening
The server is created with default settings:
const io = new Server(httpServer);By default this will:
- Accept connections from any origin.
- Attach to the root path (
/socket.io) which may collide with other middleware.
Please restrict origins and explicitly set the path, e.g.:
-const io = new Server(httpServer);
+const io = new Server(httpServer, {
+ path: "/ws",
+ cors: {
+ origin: ["https://your-ui.example.com"],
+ methods: ["GET", "POST"],
+ },
+});This blocks drive‑by websocket abuse and aligns with security best‑practices.
left a comment
There was a problem hiding this comment.
Actionable comments posted: 9
♻️ Duplicate comments (3)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.ts (1)
58-72: Consider returning a cleanup handle to avoid leaked Mongo connections
initializeMongoClientopens a long-livedMongoClientbut never exposes a way for callers (or Fastify'sonClose) to close it. In production this can leave idle sockets hanging around after hot-reloads or test runs, eventually exhausting the MongoDB connection pool.const initializeMongoClient = async function ( database: string, host: string, port: string -): Promise<Db> { +): Promise<{db: Db; close: () => Promise<void>}> { const mongoUri = `mongodb://${host}:${port}`; const mongoClient = new MongoClient(mongoUri); try { await mongoClient.connect(); - return mongoClient.db(database); + return { + db: mongoClient.db(database), + close: () => mongoClient.close(), + }; } catch (e) { throw new Error("MongoDB connection error", {cause: e}); } };Fastify plugins can then register:
fastify.addHook("onClose", async () => { await mongoHandle.close(); });components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts (2)
116-124: Socket.IO initialisation lacks CORS and path hardeningThe server is created with default settings, which could expose your application to security risks.
this.#io = new Server< ClientToServerEvents, ServerToClientEvents, SocketData ->(fastify.server); +>(fastify.server, { + path: "/ws", + cors: { + origin: ["https://your-ui.example.com"], + methods: ["GET", "POST"], + }, +});This blocks drive-by websocket abuse and aligns with security best practices.
342-378: N listeners per watcher ⇒ N² eventsEach new subscriber attaches its own
changelistener to the sameChangeStream. When many clients share a query, multiple listeners fire per change, which can impact performance.Move the listener creation into
MongoServerCollection.getWatcherwhen the watcher is first created, and inside it broadcast to the room derived fromqueryHash. Only one listener per watcher will keep event density O(1).
🧹 Nitpick comments (6)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts (4)
221-222: Use positive condition for better readabilityPrefer positive conditions over negated ones for better readability.
-if (false === collections?.includes(collectionName)) { +if (!collections?.includes(collectionName)) { collections.push(collectionName); }
292-293: Use positive condition for better readabilityPrefer positive conditions over negated ones for better readability.
-if (false === queries?.includes(queryId)) { +if (!queries?.includes(queryId)) { queries.push(queryId); }
298-300: Use positive condition for better readabilityPrefer positive conditions over negated ones for better readability. This pattern appears throughout the code.
-if (false === this.#queryIdToCollectionNameMap.has(queryId)) { +if (!this.#queryIdToCollectionNameMap.has(queryId)) { this.#queryIdToCollectionNameMap.set(queryId, collectionName); }
403-404: Use positive condition for better readabilityThis follows the same pattern. Consider updating all similar instances in the file for consistency.
-if (false === queryIds.includes(queryId)) { +if (!queryIds.includes(queryId)) { return; }components/log-viewer-webui/server/src/plugins/FastifyMongoServer/MongoServerCollection.ts (2)
119-149: Improve throttled update mechanismThe current implementation uses a mix of direct emission and timeout-based throttling which could be simplified. Also, the error is logged to console instead of using a proper logger.
Here's a more concise implementation using a throttling mechanism:
const emitUpdate = async () => { - const currentTime = Date.now(); - - if (updateTimeout <= currentTime - lastEmitTime) { - lastEmitTime = currentTime; - this.io.to(`${queryId}`).emit("collection::find::update", { - queryId: queryId, - data: await this.collection.find(query, options).toArray(), - }); - - return; - } - if (!pendingUpdate) { pendingUpdate = true; - // eslint-disable-next-line @typescript-eslint/no-misused-promises - setTimeout(async () => { + const timeToNextUpdate = Math.max(0, (lastEmitTime + updateTimeout) - Date.now()); + setTimeout(() => { + void (async () => { + lastEmitTime = Date.now(); + try { + const data = await this.collection.find(query, options).toArray(); + this.io.to(`${queryId}`).emit("collection::find::update", { + queryId: queryId, + data: data, + }); + } catch (error) { + // Use proper logger instead of console + console.error("Error fetching data for update:", error); + } finally { + pendingUpdate = false; + } + })(); + }, timeToNextUpdate); + } +};
173-177: Apply consistent style for conditional checksTo maintain consistency with the code style used elsewhere in the project:
-if (1 < watcher.subscribers.length) { +if (watcher.subscribers.length > 1) { // Remove the connectionId from the subscribers list watcher.subscribers = watcher.subscribers.filter((id) => id !== connectionId); removed = false;
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/MongoServerCollection.ts(1 hunks)components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts(1 hunks)components/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.ts(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.tscomponents/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.tscomponents/log-viewer-webui/server/src/plugins/FastifyMongoServer/MongoServerCollection.ts
🧬 Code Graph Analysis (2)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.ts (1)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts (4)
collectionName(207-210)collectionName(289-301)collectionName(312-330)collectionName(396-425)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/MongoServerCollection.ts (2)
components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts (4)
collectionName(207-210)collectionName(289-301)collectionName(312-330)collectionName(396-425)components/log-viewer-webui/server/src/plugins/FastifyMongoServer/utils.ts (1)
convertFindToChangeStreamQuery(74-74)
| * TODO: Improve this? Think about security (other queries should not be able to kick others | ||
| * offline; maybe add a ref count then), performance, and collision chances. | ||
| * | ||
| * Generates a unique hash for a given query and options. | ||
| * This hash is used to identify and manage change streams for specific queries. | ||
| * | ||
| * @param collectionName | ||
| * @param query The query object to be hashed. | ||
| * @param options The options object to be hashed. | ||
| * @return A string representing the unique hash for the query and options. | ||
| */ |
There was a problem hiding this comment.
🛠️ Refactor suggestion
TODO comment needs to be addressed
The TODO comment indicates that the query hash generation needs improvement for security, performance, and collision avoidance.
Consider implementing a more robust solution for generating query hashes:
-// eslint-disable-next-line no-warning-comments
-TODO: Improve this? Think about security (other queries should not be able to kick others
- offline; maybe add a ref count then), performance, and collision chances.
-
/**
* Generates a unique hash for a given query and options.
* This hash is used to identify and manage change streams for specific queries.
*
* @param collectionName
* @param query The query object to be hashed.
* @param options The options object to be hashed.
* @return A string representing the unique hash for the query and options.
*/
const getQueryHash = function (
collectionName: string,
query: object,
options: object
): string {
- return JSON.stringify({collectionName, query, options});
+ // Use a deterministic JSON stringifier to handle key order consistency
+ const stringifyStable = (obj: object): string => {
+ const allKeys: string[] = [];
+ const seen = new WeakMap();
+
+ // Collect all keys
+ JSON.stringify(obj, (key, value) => {
+ if (typeof value === 'object' && value !== null) {
+ if (seen.has(value)) {
+ return '[Circular]';
+ }
+ seen.set(value, true);
+ allKeys.push(...Object.keys(value));
+ }
+ return value;
+ });
+
+ // Create a stable version with sorted keys
+ return JSON.stringify(obj, allKeys.sort());
+ };
+
+ return stringifyStable({collectionName, query, options});
};📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| * TODO: Improve this? Think about security (other queries should not be able to kick others | |
| * offline; maybe add a ref count then), performance, and collision chances. | |
| * | |
| * Generates a unique hash for a given query and options. | |
| * This hash is used to identify and manage change streams for specific queries. | |
| * | |
| * @param collectionName | |
| * @param query The query object to be hashed. | |
| * @param options The options object to be hashed. | |
| * @return A string representing the unique hash for the query and options. | |
| */ | |
| /** | |
| * Generates a unique hash for a given query and options. | |
| * | |
| * @param collectionName | |
| * @param query The query object to be hashed. | |
| * @param options The options object to be hashed. | |
| * @return A string representing the unique hash for the query and options. | |
| */ | |
| const getQueryHash = function ( | |
| collectionName: string, | |
| query: object, | |
| options: object | |
| ): string { | |
| // Use a deterministic JSON stringifier to handle key order consistency | |
| const stringifyStable = (obj: object): string => { | |
| const allKeys: string[] = []; | |
| const seen = new WeakMap(); | |
| // Collect all keys | |
| JSON.stringify(obj, (key, value) => { | |
| if (typeof value === 'object' && value !== null) { | |
| if (seen.has(value)) { | |
| return '[Circular]'; | |
| } | |
| seen.set(value, true); | |
| allKeys.push(...Object.keys(value)); | |
| } | |
| return value; | |
| }); | |
| // Create a stable version with sorted keys | |
| return JSON.stringify(obj, allKeys.sort()); | |
| }; | |
| return stringifyStable({ collectionName, query, options }); | |
| }; |
| this.#fastify.log | ||
| .error(`Error checking collection existence: | ||
| ${collectionName} - ${(error as Error).error}`); | ||
| callback({ |
There was a problem hiding this comment.
Fix error handling and string formatting
The error message has template literal syntax issues causing incorrect formatting and potentially exposing internal error information to clients.
this.#fastify.log
- .error(`Error checking collection existence:
- ${collectionName} - ${(error as Error).error}`);
+ .error(`Error checking collection existence: ${collectionName}`, error);
callback({
collectionName: collectionName,
error: "An error occurred while checking the collection.",
});📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| this.#fastify.log | |
| .error(`Error checking collection existence: | |
| ${collectionName} - ${(error as Error).error}`); | |
| callback({ | |
| this.#fastify.log | |
| .error(`Error checking collection existence: ${collectionName}`, error); | |
| callback({ | |
| collectionName: collectionName, | |
| error: "An error occurred while checking the collection.", | |
| }); |
| this.#io = new Server< | ||
| ClientToServerEvents, | ||
| ServerToClientEvents, | ||
| SocketData | ||
| >(fastify.server); | ||
|
|
||
| this.#startSocketListeners(); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Enhance collection management with strong error handling
Make sure you retain a reference to the MongoDB client to enable proper cleanup when Fastify is shutting down. You'll need to update the static create method and MongoDB client initialization.
static async create (
fastify: FastifyInstance,
database: string,
host: string,
port: number
): Promise<MongoSocketIoServer> {
- const mongoDb = await initializeMongoClient(
+ const mongoClient = await initializeMongoClient(
database,
host,
port.toString()
);
- return new MongoSocketIoServer(fastify, mongoDb);
+ const instance = new MongoSocketIoServer(fastify, mongoClient.db);
+
+ // Register cleanup hook to close MongoDB connection when Fastify shuts down
+ fastify.addHook("onClose", async () => {
+ await mongoClient.close();
+ });
+
+ return instance;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| this.#io = new Server< | |
| ClientToServerEvents, | |
| ServerToClientEvents, | |
| SocketData | |
| >(fastify.server); | |
| this.#startSocketListeners(); | |
| } | |
| static async create ( | |
| fastify: FastifyInstance, | |
| database: string, | |
| host: string, | |
| port: number | |
| ): Promise<MongoSocketIoServer> { | |
| - const mongoDb = await initializeMongoClient( | |
| + const mongoClient = await initializeMongoClient( | |
| database, | |
| host, | |
| port.toString() | |
| ); | |
| - return new MongoSocketIoServer(fastify, mongoDb); | |
| + const instance = new MongoSocketIoServer(fastify, mongoClient.db); | |
| + | |
| + // Register cleanup hook to close MongoDB connection when Fastify shuts down | |
| + fastify.addHook("onClose", async () => { | |
| + await mongoClient.close(); | |
| + }); | |
| + | |
| + return instance; | |
| } |
| console.error(`Error closing watcher for queryId ${queryId}:`, err); | ||
| }); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Use proper logging instead of console.error
Replace console logging with the provided logging mechanism for better consistency and control over log levels.
-console.error(`Error closing watcher for queryId ${queryId}:`, err);
+// Pass io server's logger or inject a logger dependency in constructor
+this.logger.error(`Error closing watcher for queryId ${queryId}:`, err);Committable suggestion skipped: line range outside the PR's diff.
| console.warn(`No watcher found for queryId ${queryId}`); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Use proper logging instead of console.warn
Replace console logging with the provided logging mechanism for better consistency and control over log levels.
-console.warn(`No watcher found for queryId ${queryId}`);
+// Pass io server's logger or inject a logger dependency in constructor
+this.logger.warn(`No watcher found for queryId ${queryId}`);Committable suggestion skipped: line range outside the PR's diff.
| console.warn("Attempted to remove reference when count is already 0"); | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Use proper logging instead of console.warn
Replace console logging with the provided logging mechanism for better consistency and control over log levels.
-console.warn("Attempted to remove reference when count is already 0");
+// Pass io server's logger or inject a logger dependency in constructor
+this.logger.warn("Attempted to remove reference when count is already 0");Committable suggestion skipped: line range outside the PR's diff.
| console.error("Error fetching data for final update:", error); | ||
| } finally { |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Use proper logging instead of console.error
Replace console logging with the provided logging mechanism for better consistency and control over log levels.
-console.error("Error fetching data for final update:", error);
+// Pass io server's logger or inject a logger dependency in constructor
+this.logger.error("Error fetching data for final update:", error);Committable suggestion skipped: line range outside the PR's diff.
| // eslint-disable-next-line @typescript-eslint/no-misused-promises | ||
| watcher.changeStream.on("change", emitUpdate); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Improve change stream listener efficiency
The current implementation adds a unique change listener for each socket connection. For many subscribers to the same query, this results in multiple redundant executions of the same query.
Refactor the watcher management to ensure only one change listener per change stream, regardless of the number of subscribers:
-// eslint-disable-next-line @typescript-eslint/no-misused-promises
-watcher.changeStream.on("change", emitUpdate);
+// Only add the change listener once when watcher is created
+if (watcher.subscribers.length === 1) {
+ // eslint-disable-next-line @typescript-eslint/no-misused-promises
+ watcher.changeStream.on("change", emitUpdate);
+}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // eslint-disable-next-line @typescript-eslint/no-misused-promises | |
| watcher.changeStream.on("change", emitUpdate); | |
| // Only add the change listener once when watcher is created | |
| if (watcher.subscribers.length === 1) { | |
| // eslint-disable-next-line @typescript-eslint/no-misused-promises | |
| watcher.changeStream.on("change", emitUpdate); | |
| } |
| class MongoServerCollection { | ||
| // Reference count for active subscriptions | ||
| private count: number; | ||
|
|
||
| // MongoDB collection instance | ||
| private collection: Collection; | ||
|
|
||
| private io: Server; | ||
|
|
||
| // Map of active change streams keyed by queryId | ||
| private watchers: Map<number, Watcher> = new Map(); | ||
|
|
||
| /** | ||
| * Creates an instance of MongoReplicaServerCollection. | ||
| * | ||
| * @param collectionName The name of the collection to manage. | ||
| * @param io The Socket.IO server instance. | ||
| * @param mongoDb The MongoDB database instance. | ||
| */ | ||
| constructor (collectionName: string, io: Server, mongoDb: Db) { | ||
| this.count = 0; | ||
| this.collection = mongoDb.collection(collectionName); | ||
| this.io = io; | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add proper logging support
The class needs a logger for consistent log handling rather than using console statements. Consider adding a logger dependency to the constructor.
class MongoServerCollection {
// Reference count for active subscriptions
private count: number;
// MongoDB collection instance
private collection: Collection;
private io: Server;
+
+ // Logger for consistent logging
+ private logger: any; // Use appropriate logger interface type here
// Map of active change streams keyed by queryId
private watchers: Map<number, Watcher> = new Map();
/**
* Creates an instance of MongoReplicaServerCollection.
*
* @param collectionName The name of the collection to manage.
* @param io The Socket.IO server instance.
* @param mongoDb The MongoDB database instance.
+ * @param logger The logger instance.
*/
- constructor (collectionName: string, io: Server, mongoDb: Db) {
+ constructor (collectionName: string, io: Server, mongoDb: Db, logger: any) {
this.count = 0;
this.collection = mongoDb.collection(collectionName);
this.io = io;
+ this.logger = logger;
}Then, update the call site in FastifyMongoServer/index.ts to pass the Fastify logger:
// In index.ts
collection = new MongoServerCollection(collectionName, this.#io, this.#mongoDb);
// becomes:
collection = new MongoServerCollection(collectionName, this.#io, this.#mongoDb, this.#fastify.log);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| class MongoServerCollection { | |
| // Reference count for active subscriptions | |
| private count: number; | |
| // MongoDB collection instance | |
| private collection: Collection; | |
| private io: Server; | |
| // Map of active change streams keyed by queryId | |
| private watchers: Map<number, Watcher> = new Map(); | |
| /** | |
| * Creates an instance of MongoReplicaServerCollection. | |
| * | |
| * @param collectionName The name of the collection to manage. | |
| * @param io The Socket.IO server instance. | |
| * @param mongoDb The MongoDB database instance. | |
| */ | |
| constructor (collectionName: string, io: Server, mongoDb: Db) { | |
| this.count = 0; | |
| this.collection = mongoDb.collection(collectionName); | |
| this.io = io; | |
| } | |
| // File: components/log-viewer-webui/server/src/plugins/FastifyMongoServer/MongoServerCollection.ts | |
| class MongoServerCollection { | |
| // Reference count for active subscriptions | |
| private count: number; | |
| // MongoDB collection instance | |
| private collection: Collection; | |
| private io: Server; | |
| // Logger for consistent logging | |
| private logger: any; // Use appropriate logger interface type here | |
| // Map of active change streams keyed by queryId | |
| private watchers: Map<number, Watcher> = new Map(); | |
| /** | |
| * Creates an instance of MongoReplicaServerCollection. | |
| * | |
| * @param collectionName The name of the collection to manage. | |
| * @param io The Socket.IO server instance. | |
| * @param mongoDb The MongoDB database instance. | |
| * @param logger The logger instance. | |
| */ | |
| constructor( | |
| collectionName: string, | |
| io: Server, | |
| mongoDb: Db, | |
| logger: any | |
| ) { | |
| this.count = 0; | |
| this.collection = mongoDb.collection(collectionName); | |
| this.io = io; | |
| this.logger = logger; | |
| } | |
| // ... | |
| } |
| class MongoServerCollection { | |
| // Reference count for active subscriptions | |
| private count: number; | |
| // MongoDB collection instance | |
| private collection: Collection; | |
| private io: Server; | |
| // Map of active change streams keyed by queryId | |
| private watchers: Map<number, Watcher> = new Map(); | |
| /** | |
| * Creates an instance of MongoReplicaServerCollection. | |
| * | |
| * @param collectionName The name of the collection to manage. | |
| * @param io The Socket.IO server instance. | |
| * @param mongoDb The MongoDB database instance. | |
| */ | |
| constructor (collectionName: string, io: Server, mongoDb: Db) { | |
| this.count = 0; | |
| this.collection = mongoDb.collection(collectionName); | |
| this.io = io; | |
| } | |
| // File: components/log-viewer-webui/server/src/plugins/FastifyMongoServer/index.ts | |
| - collection = new MongoServerCollection(collectionName, this.#io, this.#mongoDb); | |
| + collection = new MongoServerCollection( | |
| + collectionName, | |
| + this.#io, | |
| + this.#mongoDb, | |
| + this.#fastify.log | |
| + ); |
commented
Apr 29, 2025
|
I added a stylistic refactor here. and some other minor logic changes - #865 |
left a comment
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (9)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/utils.ts (1)
67-76: Mongo URI construction ignores auth, SSL & options
initializeMongoClienthard-codesmongodb://host:portwhich breaks for:• replica sets / SRV records (
mongodb+srv://…)
• authentication credentials
• TLS requirementsAccept a full URI or extend
DbOptions:interface DbOptions { uri?: string; // takes precedence host: string; port: number; username?: string; password?: string; tls?: boolean; database: string; }Then build the URI accordingly, or simply
new MongoClient(options.uri ?? ...).components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts (2)
161-171: Negation style deviates from code-base guideline
else if (!emitTimeout)violates the “preferfalse == <expr>over!<expr>” rule.
Same pattern occurs in other files – worth running a linter autofix.- } else if (!emitTimeout) { + } else if (false == emitTimeout) {
80-88: Return value description & implementation out of syncThe JSDoc says the method returns “True if connection is last subscriber”, yet the early-exit path when the watcher is missing also returns
false.
Clarify documentation or distinguish between “watcher not found” and “still has subscribers” with separate return codes / an enum.components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts (2)
238-245: Negation shorthand violates code-style guideline
if (!collectionName)should follow the documented guideline:- if (!collectionName) { - this.#fastify.log.error("Collection name is undefined"); - return; - } + if (false == collectionName) { + this.#fastify.log.error("Collection name is undefined"); + return; + }
204-221:#getQueryIdperformance & collision riskIterating
Map.entries()each time scales O(N).
A reverse map (hash → id) or simply using the hash string as the key avoids linear scans and removes the max-key calculation.- const queryHash = getQueryHash(queryParams); - for (const [queryId, hash] of this.#queryIdtoQueryHashMap.entries()) { - if (hash === queryHash) { - return queryId; - } - } - let queryId = 0; - ... - this.#queryIdtoQueryHashMap.set(queryId, queryHash); - return queryId; + const hash = getQueryHash(queryParams); + let id = this.#queryIdtoQueryHashMap.get(hash); + if ("undefined" !== typeof id) { + return id; + } + id = this.#queryIdtoQueryHashMap.size; + this.#queryIdtoQueryHashMap.set(hash, id); + return id;components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts (4)
33-58: Add JSDoc for each client-to-server event
The top-level comment is helpful, but adding small JSDoc blocks per event (e.g., for"disconnect","collection::init", etc.) will improve discoverability and make it clearer what each callback’s arguments and expected behavior are.
41-46: SimplifyResponsegeneric for find-toArray callback
UsingResponse<{data: Document[]}>leads to payloads shaped like{ data: { data: Document[] } }. Instead, consider:"collection::find::toArray": ( args: { query: Filter<Document>; options: FindOptions }, callback: (res: Response<Document[]>) => void ) => void;so that
Success<Document[]>yields{ data: Document[] }.
80-82: Replace empty interface with a type alias
An empty interface is equivalent to{}. You can simplify this section by removing the ESLint disable and using:type InterServerEvents = {};This aligns with Biome’s recommendation and reduces boilerplate.
🧰 Tools
🪛 Biome (1.9.4)
[error] 80-82: An empty interface is equivalent to {}.
Safe fix: Use a type alias instead.
(lint/suspicious/noEmptyInterface)
128-128: Consider making the client update timeout configurable
The hardcodedCLIENT_UPDATE_TIMEOUT_MS = 500may need tuning in different environments. Exposing this as a plugin option or via an environment variable will let operators adjust throttling without code changes.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
components/log-viewer-webui/server/src/app.ts(2 hunks)components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts(1 hunks)components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts(1 hunks)components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts(1 hunks)components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/utils.ts(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- components/log-viewer-webui/server/src/app.ts
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/utils.tscomponents/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.tscomponents/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.tscomponents/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts
🪛 Biome (1.9.4)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts
[error] 80-82: An empty interface is equivalent to {}.
Safe fix: Use a type alias instead.
(lint/suspicious/noEmptyInterface)
🔇 Additional comments (2)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/typings.ts (2)
1-6: Well-scoped imports
Imports frommongodbare grouped together and theSocketimport fromsocket.iois clearly separated. This makes it easy to see third-party dependencies at a glance.
7-7: Verify Socket.IO type definitions are installed
You’re importingSocketfrom"socket.io". Ensure that the corresponding TypeScript definitions (@types/socket.ioor the built-in types in your Socket.IO version) are present indevDependencies, otherwise the build may fail with missing type declarations.
| const getQueryHash = function ( | ||
| queryParams: QueryParameters, | ||
| ): string { | ||
| return JSON.stringify(queryParams); | ||
| }; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
JSON.stringify is order-sensitive → equal queries may receive different hashes
Using JSON.stringify directly means {a:1, b:2} and {b:2, a:1} hash differently even though they are semantically identical.
A stable, canonical serialiser (e.g. canonical-json, fast-json-stable-stringify, or a custom “alphabetise keys” helper) will prevent duplicate watchers and wasted sockets.
-import {type QueryParameters} from "./typings.js";
+import {type QueryParameters} from "./typings.js";
+import stringify from "fast-json-stable-stringify"; // ← tiny, zero-dep
...
- return JSON.stringify(queryParams);
+ return stringify(queryParams);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const getQueryHash = function ( | |
| queryParams: QueryParameters, | |
| ): string { | |
| return JSON.stringify(queryParams); | |
| }; | |
| import {type QueryParameters} from "./typings.js"; | |
| import stringify from "fast-json-stable-stringify"; // ← tiny, zero-dep | |
| const getQueryHash = function ( | |
| queryParams: QueryParameters, | |
| ): string { | |
| return stringify(queryParams); | |
| }; |
| 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; | ||
| }; |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Edge cases for nested & operator-based queries are not handled
convertQueryToChangeStreamFormat works only for simple top-level key/value pairs.
Queries that contain operators like $and, $or, or nested documents ({ "meta.type": "foo" }) will not be rewritten correctly, which means the change-stream will miss updates for those queries.
Consider a recursive approach that walks the query tree and rewrites only leaf field paths, or document this limitation explicitly so that callers do the conversion beforehand.
| async #collectionInitListener ( | ||
| socket: MongoCustomSocket, | ||
| requestArgs: {collectionName: string}, | ||
| callback:(res: Response<void>) => void | ||
| ): Promise<void> { | ||
| const {collectionName} = requestArgs; | ||
| this.#fastify.log.info( | ||
| `Socket ${socket.id} requested init of collection ${collectionName}` | ||
| ); | ||
|
|
||
| const exists = await this.#collectionExists(collectionName); | ||
| if (false === exists) { | ||
| this.#fastify.log.error(`Collection ${collectionName} does not exist in MongoDB`); | ||
| callback({ | ||
| error: `Collection ${collectionName} does not exist in MongoDB`, | ||
| }); | ||
|
|
||
| return; | ||
| } | ||
|
|
||
|
|
||
| socket.data.collectionName = collectionName; | ||
| } |
There was a problem hiding this comment.
Success path never invokes callback → client hangs
#collectionInitListener calls the callback only on the error branch, leaving the promise unresolved for happy-path initialisation.
Clients waiting on the acknowledgement will time out.
@@
- socket.data.collectionName = collectionName;
+ socket.data.collectionName = collectionName;
+ callback({}); // acknowledge success📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async #collectionInitListener ( | |
| socket: MongoCustomSocket, | |
| requestArgs: {collectionName: string}, | |
| callback:(res: Response<void>) => void | |
| ): Promise<void> { | |
| const {collectionName} = requestArgs; | |
| this.#fastify.log.info( | |
| `Socket ${socket.id} requested init of collection ${collectionName}` | |
| ); | |
| const exists = await this.#collectionExists(collectionName); | |
| if (false === exists) { | |
| this.#fastify.log.error(`Collection ${collectionName} does not exist in MongoDB`); | |
| callback({ | |
| error: `Collection ${collectionName} does not exist in MongoDB`, | |
| }); | |
| return; | |
| } | |
| socket.data.collectionName = collectionName; | |
| } | |
| async #collectionInitListener ( | |
| socket: MongoCustomSocket, | |
| requestArgs: {collectionName: string}, | |
| callback: (res: Response<void>) => void | |
| ): Promise<void> { | |
| const {collectionName} = requestArgs; | |
| this.#fastify.log.info( | |
| `Socket ${socket.id} requested init of collection ${collectionName}` | |
| ); | |
| const exists = await this.#collectionExists(collectionName); | |
| if (false === exists) { | |
| this.#fastify.log.error(`Collection ${collectionName} does not exist in MongoDB`); | |
| callback({ | |
| error: `Collection ${collectionName} does not exist in MongoDB`, | |
| }); | |
| return; | |
| } | |
| socket.data.collectionName = collectionName; | |
| callback({}); // acknowledge success | |
| } |
left a comment
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts (1)
153-178: Success path never invokescallback→ client hangsThe
#collectionInitListenercalls the callback only on the error branch, leaving the promise unresolved for happy-path initialization. Clients waiting on the acknowledgement will time out.- socket.data.collectionName = collectionName; + socket.data.collectionName = collectionName; + callback({}); // acknowledge success
🧹 Nitpick comments (2)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts (2)
243-247: Use explicit comparison instead of negation operatorAccording to the coding guidelines, we should prefer
false == <expression>rather than!<expression>.- if (!collectionName) { + if (false === collectionName || undefined === collectionName) {
214-221: Simplify query ID generationThe query ID generation logic could be simplified by using a class-level counter that increments for each new query.
- let queryId = 0; - if (0 === this.#queryIdtoQueryHashMap.size) { - this.#queryIdtoQueryHashMap.set(queryId, queryHash); - } else { - const maxKey = Math.max(...Array.from(this.#queryIdtoQueryHashMap.keys())); - queryId = maxKey + 1; - this.#queryIdtoQueryHashMap.set(queryId, queryHash); - } + // Use a class property to track the next query ID + if (!this.nextQueryId) { + this.nextQueryId = 0; + } + const queryId = this.nextQueryId++; + this.#queryIdtoQueryHashMap.set(queryId, queryHash);
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts
🔇 Additional comments (1)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts (1)
39-44: Consider client-side socket sharing per PR commentsBased on PR comments, consider implementing socket sharing on the client side using a singleton pattern rather than creating separate socket connections for each collection. This would align with browser limits on WebSocket connections per domain.
You could enhance the design by:
- Using a shared socket on the client side
- Including query hash in the event payload instead of using different event names
- Leveraging socket.io rooms for efficient resource sharing
| const queryId = this.#getQueryId(queryParameters); | ||
| await collection.getWatcher(queryParameters, queryId, socket); | ||
| callback({data: {queryId}}); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add error handling for getWatcher
The call to collection.getWatcher() might fail, but there's no error handling here. Add a try/catch block to gracefully handle potential errors and provide helpful feedback to the client.
- const queryId = this.#getQueryId(queryParameters);
- await collection.getWatcher(queryParameters, queryId, socket);
- callback({data: {queryId}});
+ const queryId = this.#getQueryId(queryParameters);
+ try {
+ await collection.getWatcher(queryParameters, queryId, socket);
+ callback({data: {queryId}});
+ } catch (error) {
+ this.#fastify.log.error(`Error getting watcher: ${error}`);
+ callback({
+ error: "Failed to subscribe to collection updates",
+ });
+ return;
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const queryId = this.#getQueryId(queryParameters); | |
| await collection.getWatcher(queryParameters, queryId, socket); | |
| callback({data: {queryId}}); | |
| const queryId = this.#getQueryId(queryParameters); | |
| try { | |
| await collection.getWatcher(queryParameters, queryId, socket); | |
| callback({ data: { queryId } }); | |
| } catch (error) { | |
| this.#fastify.log.error(`Error getting watcher: ${error}`); | |
| callback({ | |
| error: "Failed to subscribe to collection updates", | |
| }); | |
| return; | |
| } |
left a comment
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (2)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts (2)
153-178: Success path never invokescallback→ client hangs
#collectionInitListenercalls the callback only on the error branch, leaving the promise unresolved for happy-path initialization. Clients waiting on the acknowledgement will time out.if (socket.disconnected) { return; } socket.data.collectionName = collectionName; +callback({}); // acknowledge success
264-266: Add error handling for getWatcherThe call to
collection.getWatcher()might fail, but there's no error handling here. Add a try/catch block to gracefully handle potential errors and provide helpful feedback to the client.const queryId = this.#getQueryId(queryParameters); - await collection.getWatcher(queryParameters, queryId, socket); - callback({data: {queryId}}); + try { + await collection.getWatcher(queryParameters, queryId, socket); + callback({data: {queryId}}); + } catch (error) { + this.#fastify.log.error(`Error getting watcher: ${error}`); + callback({ + error: "Failed to subscribe to collection updates", + }); + return; + }
🧹 Nitpick comments (6)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts (6)
1-3: TODO comments should be addressed before mergingThe file starts with TODOs that should be addressed. Moving listeners to separate files would improve code organization and maintainability.
Consider creating separate files for the different socket event listeners to reduce the complexity of this class, as indicated by your TODO comment.
243-247: Use the coding standard format for boolean expressionsAccording to the project guidelines, boolean negations should use the format
false == expressionrather than!expression.- if (!collectionName) { + if (false == collectionName) {
327-328: Combine boolean conditions for better readabilityThe combination of
"undefined" === typeof subscribedQueryIdsandfalse === subscribedQueryIds.includes(queryId)can be simplified.- if ("undefined" === typeof subscribedQueryIds || - false === subscribedQueryIds.includes(queryId) - ) { + if (!subscribedQueryIds || false === subscribedQueryIds.includes(queryId)) {
189-193: Consider using optional chaining operator for cleaner codeThe check for undefined followed by setting a value can be simplified using modern JavaScript/TypeScript features.
- const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socketId); - if ("undefined" === typeof subscribedQueryIds) { - this.#subscribedQueryIdsMap.set(socketId, [queryId]); - - return; - } + const subscribedQueryIds = this.#subscribedQueryIdsMap.get(socketId) || []; + if (false === subscribedQueryIds.includes(queryId)) { + subscribedQueryIds.push(queryId); + } + this.#subscribedQueryIdsMap.set(socketId, subscribedQueryIds);
214-224: Simplify query ID generation logicThe current implementation has conditional logic that can be simplified. Using
Math.max()with a default value would make this more concise.- let queryId = 0; - if (0 === this.#queryIdtoQueryHashMap.size) { - this.#queryIdtoQueryHashMap.set(queryId, queryHash); - } else { - const maxKey = Math.max(...Array.from(this.#queryIdtoQueryHashMap.keys())); - queryId = maxKey + 1; - this.#queryIdtoQueryHashMap.set(queryId, queryHash); - } + const queryId = this.#queryIdtoQueryHashMap.size === 0 ? 0 : + Math.max(...Array.from(this.#queryIdtoQueryHashMap.keys())) + 1; + this.#queryIdtoQueryHashMap.set(queryId, queryHash);
337-340: Use inline filter for more concise codeThe current implementation sets a filtered array back to the map in multiple steps. This can be done more concisely.
- this.#subscribedQueryIdsMap.set( - socket.id, - subscribedQueryIds.filter((id) => id !== queryId) - ); + const filteredIds = subscribedQueryIds.filter((id) => id !== queryId); + if (filteredIds.length > 0) { + this.#subscribedQueryIdsMap.set(socket.id, filteredIds); + } else { + this.#subscribedQueryIdsMap.delete(socket.id); + }This also handles cleanup if no subscriptions remain for this socket.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts(1 hunks)components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/MongoWatcherCollection.ts
🧰 Additional context used
📓 Path-based instructions (1)
`**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}`: - Prefer `false == ` rather than `!`.
**/*.{cpp,hpp,java,js,jsx,tpp,ts,tsx}: - Preferfalse == <expression>rather than!<expression>.
components/log-viewer-webui/server/src/plugins/MongoSocketIoServer/index.ts
|
|
||
| /** | ||
| * Manages client interactions with MongoDB. | ||
| * | ||
| * TODO: In current implementation, all queries in a collection are sent using the same event. A | ||
| * potential improvement would be to use different event names per query. | ||
| */ |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Apply room-based optimization mentioned in PR comments
The PR comments suggest using socket.io rooms to optimize watcher sharing across multiple clients. Consider implementing this improvement as mentioned in the discussion.
The current TODO comment addresses only part of the optimization. Consider implementing socket.io rooms keyed by query hashes, as discussed in the PR comments. This would allow sharing a single watcher instance across multiple clients subscribing to the same query, improving resource usage.
There was a problem hiding this comment.
For now we will do our own registration, since https://www.npmjs.com/package/fastify-socket.io does not support Fastify v5 or above.
| import { | ||
| Collection, | ||
| Db, | ||
| } from "mongodb"; |
There was a problem hiding this comment.
Can we write
| import { | |
| Collection, | |
| Db, | |
| } from "mongodb"; | |
| import type { | |
| Collection, | |
| Db, | |
| } from "mongodb"; |
| * Provides watchers for MongoDB queries to a specific collection. | ||
| */ | ||
| class MongoWatcherCollection { | ||
| private collection: Collection; |
There was a problem hiding this comment.
Can we use the #collection sytax instead of the private keyword?
| class MongoWatcherCollection { | ||
| private collection: Collection; | ||
|
|
||
| private io: Server; |
There was a problem hiding this comment.
ditto: let's use the # syntax if there's no technical restriction.
| private io: Server; | ||
|
|
||
| // Active watchers | ||
| private queryIdtoWatcherMap: Map<QueryId, Watcher> = new Map(); |
| Collection, | ||
| Db, | ||
| } from "mongodb"; | ||
| import {Server} from "socket.io"; |
There was a problem hiding this comment.
Similarly, if we're only using the Server as a type:
| import {Server} from "socket.io"; | |
| import type {Server} from "socket.io"; |
| * @param io | ||
| * @param mongoDb | ||
| */ | ||
| constructor (collectionName: string, io: Server, mongoDb: Db) { |
There was a problem hiding this comment.
This is really nit-picking but I find it counter intuitive to list the arguments io and mongoDb after collectionName.
Accepting collectionName and connectionOptions would have made sense. That said, for the sake of simplicity, can we use a single object for the constructor arguments? e.g.,
constructor ({collectionName, io, mongoDb}: {collectionName: string, io: Server, mongoDb: Db})
Please see the below comment for interface update suggestions.
| * @param queryId | ||
| * @param socket | ||
| */ | ||
| async getWatcher ( |
There was a problem hiding this comment.
This is an unusual design choice in the interface. The getWatcher method is mixing two very different responsibilities:
- Creating/retrieving a watcher (resource management)
- Subscribing a specific socket and sending initial data (communication)
This violates the single responsibility principle and creates an awkward interface that tightly couples distinct operations.
Can we implement those interfaces instead:
interface WatcherOptions<T> {
onUpdate: (data: T[]) => void;
onError: (error: Error) => void;
debounceMillis: number;
}
class MongoWatcherCollection {
constructor(mongoDb: Db, collectionName: string) {}
async createOrGetWatcher<T>(queryId: QueryId, options: WatcherOptions<T>) {
...
return watcher;
}
addSubscriber(queryId: QueryId, socketId: string) {}
removeSubscriber (queryId: number, socketId: string): boolean {}
// add this, or maybe we can simply extend MongoWatcherCollection from mongo.Collection
find(...) {
this.#collection.find(...);
}
}
There was a problem hiding this comment.
I took inspiration from this in the new PR. My implementation is not exactly the same. But uses the same idea
commented
May 2, 2025
|
It seems there're some conflicts with |
|
For the PR title, how about: |
commented
May 5, 2025
|
This PR was moved to #880 and can be closed |
commented
May 8, 2025
|
replaced by #880 |
3 participants
Description
Server for new framework built on MongoCDC using socket.io meant to replace our current usage of meteor's publish-subscribe framework.
First PR to split #809
TODO: A few listing errors left to solve
Checklist
breaking change.
Validation performed
Manually added collection to mongo replica and tested to see that the contest of the collection could be requested through the client from the server.
Summary by CodeRabbit
New Features
Chores