feat(sse-module): server sent event - #2790
Open
Junjiequan wants to merge 25 commits into
Open
Conversation
Junjiequan
force-pushed
the
server-sent-event
branch
from
June 23, 2026 11:56
f138b1c to
827c655
Compare
Junjiequan
marked this pull request as ready for review
June 23, 2026 11:59
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- The JWT strategy now accepts tokens from a
tokenquery parameter for all routes; consider restricting this extractor to the SSE endpoints (or making it configurable) to avoid unintentionally broadening the attack surface for every authenticated endpoint. - In
SseController.stream,SseService.getEventsalready yieldsMessageEventobjects but you then wrap them again as{ data: payload }; either return the observable directly or changeSseService.emitto work with raw payloads so you don’t end up nesting MessageEvents. - In
SseService,accessGroupsis left without an explicit type or default and then accessed with optional chaining; it would be clearer and safer to type it asAccessGroupsType | undefinedand/or initialize it to a sensible default (e.g.{ admin: [] }) to avoid relying on undefined checks throughout the access logic.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The JWT strategy now accepts tokens from a `token` query parameter for all routes; consider restricting this extractor to the SSE endpoints (or making it configurable) to avoid unintentionally broadening the attack surface for every authenticated endpoint.
- In `SseController.stream`, `SseService.getEvents` already yields `MessageEvent` objects but you then wrap them again as `{ data: payload }`; either return the observable directly or change `SseService.emit` to work with raw payloads so you don’t end up nesting MessageEvents.
- In `SseService`, `accessGroups` is left without an explicit type or default and then accessed with optional chaining; it would be clearer and safer to type it as `AccessGroupsType | undefined` and/or initialize it to a sensible default (e.g. `{ admin: [] }`) to avoid relying on undefined checks throughout the access logic.
## Individual Comments
### Comment 1
<location path="src/serverSentEvent/sse.service.ts" line_range="21-24" />
<code_context>
+ string,
+ { user: JWTUser; subject: Subject<MessageEvent> }
+ >();
+ private accessGroups;
+
+ constructor(private configService: ConfigService) {
+ this.accessGroups =
+ this.configService.get<AccessGroupsType>("accessGroups");
+ }
</code_context>
<issue_to_address>
**suggestion:** Untyped `accessGroups` field can degrade type safety and hide config issues.
Because `accessGroups` is implicitly `any`, the `this.accessGroups?.admin?.includes(g)` call in `emit` is not type-checked, so issues like `admin` not being an array won’t be caught at compile time. Please type this explicitly, e.g. `private accessGroups?: AccessGroupsType;` (or a narrower type), and consider a safe default or fail-fast behavior when the config is missing or malformed.
Suggested implementation:
```typescript
private accessGroups?: AccessGroupsType;
constructor(private readonly configService: ConfigService) {
const accessGroups = this.configService.get<AccessGroupsType>('accessGroups');
if (!accessGroups) {
// Fail fast if configuration is missing. If you prefer a softer behavior,
// replace this with a default or logging-only behavior.
throw new Error('SseService: "accessGroups" configuration is missing');
}
this.accessGroups = accessGroups;
}
```
1. Ensure `AccessGroupsType` is imported into `src/serverSentEvent/sse.service.ts` (e.g. `import { AccessGroupsType } from '...';`).
2. Ensure `ConfigService` is imported from `@nestjs/config` if not already present.
3. If throwing on missing config is too strict for your environment, replace the `throw new Error(...)` with a safe default (e.g. `this.accessGroups = { admin: [] }`) and/or a logged warning.
</issue_to_address>
### Comment 2
<location path="src/serverSentEvent/sse.listener.ts" line_range="52-53" />
<code_context>
+ },
+ ]);
+
+ this.changeStream.on("change", (change) => this.onStream(change));
+ this.changeStream.on("error", (err) =>
+ Logger.error(`SSE change stream is closed due to error: ${err}`),
+ );
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Change stream error handling just logs and leaves the stream closed.
When an error occurs, the stream is only logged as closed but never restarted or cleared. After a transient MongoDB issue, the SSE listener will stop emitting updates for the rest of the process lifetime. Consider either retrying `watch` with backoff or at minimum nulling `this.changeStream` and exposing a health indicator so the process can detect and recover from an inactive stream.
Suggested implementation:
```typescript
private reconnectAttempts = 0;
private readonly maxReconnectAttempts = 5;
private readonly baseReconnectDelayMs = 1000;
onModuleInit() {
this.startChangeStream();
}
/**
* Initializes the MongoDB change stream and attaches listeners.
* Can be safely called multiple times; it will only create a new stream
* when there is no active one.
*/
private startChangeStream() {
if (this.changeStream) {
return;
}
this.changeStream = this.connection.watch([
{
$match: {
"ns.coll": { $in: Object.keys(REGISTRY) },
operationType: { $in: Object.keys(ACTION_BY_OPERATION) },
},
},
]);
this.changeStream.on("change", (change) => this.onStream(change));
this.changeStream.on("error", (err) => this.onStreamError(err));
this.reconnectAttempts = 0;
Logger.log("SSE change stream started");
}
/**
* Very simple health indicator so the process can detect an inactive stream.
* Consider wiring this into your health endpoint / readiness probe.
*/
isHealthy(): boolean {
return !!this.changeStream;
}
/**
* Handles change stream errors by closing the current stream, nulling
* the reference, and scheduling a restart with bounded backoff.
*/
private onStreamError(err: unknown) {
Logger.error(`SSE change stream is closed due to error: ${err}`);
try {
this.changeStream?.close();
} catch (closeErr) {
Logger.error(`Failed to close SSE change stream after error: ${closeErr}`);
} finally {
this.changeStream = null;
}
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
Logger.error(
`Max reconnect attempts (${this.maxReconnectAttempts}) reached for SSE change stream; will not restart automatically.`,
);
return;
}
this.reconnectAttempts += 1;
const delay =
this.baseReconnectDelayMs * Math.pow(2, this.reconnectAttempts - 1);
Logger.warn(
`Scheduling SSE change stream restart (attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts}) in ${delay}ms`,
);
setTimeout(() => this.startChangeStream(), delay);
}
```
```typescript
private onStream(change: ChangeStreamDocument) {
```
1. If your project exposes health endpoints (e.g., via `@nestjs/terminus`), you should wire `isHealthy()` into the corresponding health indicator so that an unhealthy change stream can trigger container restarts or alerts.
2. Depending on your MongoDB driver version, you may want to refine `isHealthy()` to also check `this.changeStream.closed` or similar properties if available, instead of only checking for a non-null reference.
3. If you already have configuration for retry limits/backoff, replace the hardcoded `maxReconnectAttempts` and `baseReconnectDelayMs` with values sourced from your existing config system.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
There was a problem hiding this comment.
Pull request overview
Adds a Server-Sent Events (SSE) subsystem to broadcast dataset-related events to authenticated clients, backed by MongoDB change streams, with supporting updates to auth extraction and E2E MongoDB setup.
Changes:
- Introduce global SSE module (service/controller/listener) with
/events/streamand/events/connectionsendpoints. - Enable MongoDB replica set mode in E2E docker-compose to support change streams.
- Extend JWT extraction to also accept a
tokenquery parameter (in addition to the Authorization header) and adjust mailer transport typing.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| src/serverSentEvent/sse.service.ts | Manages SSE client connections, enforces per-user connection limits, and emits access-controlled events. |
| src/serverSentEvent/sse.module.ts | Registers SSE components as a global Nest module. |
| src/serverSentEvent/sse.listener.ts | Watches MongoDB change streams and converts inserts into SSE events. |
| src/serverSentEvent/sse.controller.ts | Exposes the SSE stream endpoint and a connections monitoring endpoint. |
| src/serverSentEvent/interfaces/sse-registry.interface.ts | Defines the registry collection-name type used by the change-stream listener. |
| src/auth/strategies/jwt.strategy.ts | Adds query-parameter token extraction to JWT strategy. |
| src/app.module.ts | Wires in the SSE module and refines mailer transport typing. |
| CI/E2E/docker-compose.yaml | Enables Mongo replica set and adds a healthcheck/init logic for change streams. |
| CI/E2E/docker-compose-local.yaml | Mirrors replica set + healthcheck/init logic for local E2E environment. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+28
to
+37
| getEvents(user: JWTUser): Observable<MessageEvent> { | ||
| const userConnectionCount = [...this.clients.values()].filter( | ||
| (c) => c.user._id === user._id, | ||
| ).length; | ||
|
|
||
| if (userConnectionCount >= this.MAX_CONNECTIONS_PER_USER_PER_INSTANCE) { | ||
| throw new ForbiddenException( | ||
| `Maximum number of ${this.MAX_CONNECTIONS_PER_USER_PER_INSTANCE} open connections reached`, | ||
| ); | ||
| } |
Junjiequan
force-pushed
the
server-sent-event
branch
from
July 3, 2026 08:58
c00dfca to
204b624
Compare
Junjiequan
force-pushed
the
server-sent-event
branch
from
July 6, 2026 12:55
f96eb5d to
d7a7c05
Compare
Junjiequan
force-pushed
the
server-sent-event
branch
from
August 19, 2026 10:54
d13b2cd to
81b435e
Compare
Member
Code Review Report: server-sent-event BranchOverviewCommits AnalyzedThe
1. Logic CorrectnessThe SSE implementation is logically sound:
Potential Issues:
2. Edge Cases
3. What the Code DoesThe changes implement a Server-Sent Events (SSE) system that:
4. Assessment of ChangesThe changes make sense and follow NestJS patterns:
5. Unreachable Code
Code ChangesFiles Changed (12 files, 457 insertions, 10 deletions)
AssessmentNecessary Changes:
Questionable Changes:
Improvement Needed
VerdictPositive: The SSE implementation is well-structured, follows NestJS best practices, and correctly integrates with existing systems. The MongoDB change stream approach is appropriate for real-time notifications. Needs Work: The implementation is incomplete (only handles inserts for Datasets) and lacks test coverage. The presence of debug logging and TODOs indicates it's not production-ready. Overall: The changes are a good foundation for SSE functionality but require additional work before being merged to master. Security Review1. Potential Injection VulnerabilitiesNone identified in the SSE-specific code. However:
Verdict: No injection vulnerabilities in the new code. 2. Sensitive User Data ExposurePotential Issue:
Verdict: The connections endpoint exposes user information but is protected by authorization. However, the policy check uses a placeholder TODO comment, indicating it needs proper policy definition. 3. Insecure API UsageIssues Found:
Verdict: The query parameter token approach is a security anti-pattern, though it's limited to the 4. Authentication BypassPotential Issue:
Verdict: No direct bypass identified, but the query parameter token approach increases the attack surface. Test CoverageCurrent StateNo tests exist for the SSE module. This is a critical gap. Files Missing Tests
Suggested Tests
Security Examples1. Token in URL - Real-Life ExampleVulnerability: Bearer tokens in query parameters Example Attack: Impact:
Fix: Use only Authorization header. For EventSource API, consider:
2. Information Disclosure via Connections EndpointVulnerability: Exposing active user connections Example Attack: Impact:
Fix: Restrict the connections endpoint to admin-only access with proper policy definition. Testing for Security Use Cases1. Testing for Token in URL VulnerabilityTest File: describe('JWT Strategy - Query Token Security', () => {
it('should NOT extract token from query params for non-SSE endpoints', () => {
const req = {
path: '/api/datasets',
query: { token: 'test-token' }
} as Request;
const result = fromSseQueryAsBearerToken(req);
expect(result).toBeNull();
});
it('should extract token from query params only for /events/stream', () => {
const req = {
path: '/api/events/stream',
query: { token: 'test-token' }
} as Request;
const result = fromSseQueryAsBearerToken(req);
expect(result).toBe('test-token');
});
it('should not extract token if query param is array', () => {
const req = {
path: '/api/events/stream',
query: { token: ['token1', 'token2'] }
} as Request;
const result = fromSseQueryAsBearerToken(req);
expect(result).toBeNull();
});
});Code Affected: 2. Testing for Information DisclosureTest File: describe('SSE Controller - Connections Endpoint Security', () => {
it('should deny access to connections endpoint for non-admin users', () => {
// Setup mock user without admin privileges
const user = { _id: '1', username: 'user', currentGroups: ['pgroup1'] };
// Mock the policies guard to check policy
// Expect 403 Forbidden
});
it('should allow access to connections endpoint for admin users', () => {
// Setup mock admin user
const user = { _id: '1', username: 'admin', currentGroups: ['admin'] };
// Expect 200 OK with connections data
});
});Code Affected: 3. Testing Access Control FilteringTest File: describe('SSE Service - Access Control', () => {
it('should not send events for datasets user cannot access', () => {
const user = { _id: '1', username: 'user', currentGroups: ['pgroup1'] };
const dataset = {
ownerGroup: 'pgroup2',
accessGroups: ['pgroup3']
};
// Setup service with mock accessGroups config
// Emit event for dataset
// Verify user's subject did NOT receive the event
});
it('should send events for datasets user owns', () => {
const user = { _id: '1', username: 'user', currentGroups: ['pgroup1'] };
const dataset = {
ownerGroup: 'pgroup1',
accessGroups: []
};
// Emit event for dataset
// Verify user's subject DID receive the event
});
it('should send events for datasets in user accessGroups', () => {
const user = { _id: '1', username: 'user', currentGroups: ['pgroup1'] };
const dataset = {
ownerGroup: 'pgroup2',
accessGroups: ['pgroup1']
};
// Emit event for dataset
// Verify user's subject DID receive the event
});
});Code Affected: SummaryKey Findings
Recommendations
AttributionGenerated by Mistral Vibe. |
nitrosx
reviewed
Aug 19, 2026
nitrosx
left a comment
nitrosx
left a comment
Member
There was a problem hiding this comment.
Few minor comments.
Please also check the AI review.
About tests: should we create an additional PR with them?
Comment on lines
+11
to
+18
| const fromSseQueryAsBearerToken = (req: Request): string | null => { | ||
| if (req.path.endsWith("events/stream")) { | ||
| const token = req.query?.token; | ||
| return typeof token === "string" ? token : null; | ||
| } | ||
| return null; | ||
| }; | ||
|
|
Member
There was a problem hiding this comment.
Please check the AI review regarding passing the token on the URI.
| const jwtSecret = process.env.JWT_SECRET; | ||
| if (!jwtSecret && process.env.NODE_ENV === "production") { | ||
| throw new Error("JWT_SECRET is required in production"); | ||
| } |
Member
Author
There was a problem hiding this comment.
it's because we can simply use fallback for development environment, while for production one must set it explicitly assuming NODE_ENV is provided
- add connection removal - add connections check
…CE avoid confusion
- Introduced WatchableCollection type for MongoDB collections in SSE. - Updated SseRegistryEntry interface to include entity and DTO mapping. - Improved SseListener to handle change streams with reconnection logic. - Enhanced SseService to manage client connections and access control. - Added SseClient and SseEvent interfaces for better type safety.
Changed realTimeUpdatesEnabled from true to false.
EventSource cannot send auth headers, so mint a short-lived ticket here and pass it on the stream URL instead.
Junjiequan
force-pushed
the
server-sent-event
branch
from
August 20, 2026 16:50
501c665 to
781997f
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Adds a Server-Sent Events (SSE) module that streams MongoDB change events to authenticated clients, so the frontend gets real-time dataset notifications without polling. Works across multiple backend instance because each instance tails MongoDB's change stream independently
This feature uses MongoDB change streams, which only work on a replica set — a standalone mongod has no oplog, and change streams read from the oplog. Without --replSet, the backend's change-stream listener throws and the feature is dead.
This is a deployment-breaking change for anyone running a standalone MongoDB. A standalone instance must be converted to a replica set (single-node is enough) in order to use server sent events.
Changes:
Tests included
Documentation
official documentation info
Summary by Sourcery
Introduce a server-sent events module to stream MongoDB change events to authorized clients and expose connection monitoring endpoints.
New Features:
tokenquery parameter for compatibility with SSE connections.Enhancements:
MailerOptions["transport"]instead of a custom transport type.Build: