k1nq/TREK
1
0
Fork 0
mirror of https://github.com/mauriceboe/TREK.git synced 2026-10-18 21:16:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Serve /ws from a Nest gateway with its own adapter

Browse Source 
The transport was a hand-rolled setupWebSocket(server) that index.ts kicked
off with a dynamic import after listen(). It is a @WebSocketGateway now, with
DatabaseService and EphemeralTokenService injected instead of imported, and
the heartbeat owned by a lifecycle hook rather than an interval nobody stops.

The wire protocol does not move. Nest's WsAdapter dispatches on
{ event, data } and every deployed client sends { type, tripId }, so
TrekWsAdapter translates rather than asking clients to change. It also builds
the ws server itself instead of delegating: the base's port-0 branch uses
noServer and performs the upgrade from its own registry, which never consults
verifyClient, so delegating would have dropped the origin check and answered
101 to anyone. The per-socket error listener stays attached first, because an
unhandled one kills the process (#1576), and the flood guard counts every
inbound frame before parsing, because malformed frames are exactly the ones
worth limiting.

Four things had to stay where they were, and each is a silent failure if it
does not:

The socket registry stays module state. Seventeen bridges and
notifications.instance build `new RealtimeService()` outside the container;
rooms on a provider instance would leave their broadcasts going to an empty
map, with no error and no log.

The socket id stays a monotonic integer. The client echoes it as X-Socket-Id
and broadcast excludes the originator with Number(excludeSid). A uuid is NaN,
NaN === NaN is false, and every client would receive its own writes back,
which shows up as drag-and-drop that jumps rather than as an error.

buildApp creates the http server and binds the adapter before app.init(), and
callers take it from getHttpServer(). Nest binds gateways during init, so a
later adapter binds nothing; an adapter given the app binds to Nest's own
internal server, which this process never listens on. Both boot green with no
client able to connect.

RealtimeGatewayModule is imported by AppModule alone. RealtimeModule is
@Global, so a gateway there follows into every hand-assembled TestingModule,
and Nest's SocketModule then loads an adapter none of them set: it reaches for
@nestjs/platform-socket.io and calls process.exit(1). Eighteen e2e harnesses
died that way before the split.

src/websocket.ts survives as a re-export of the three fan-out primitives. 115
test files vi.mock that path to assert on broadcasts.

The two transport suites pass unchanged, and 30 new cases pin the pieces
directly: the pv and MFA gates, the numeric socket id, the plugin announcement
firing before the room check, both sides of the origin allowlist, and the
flood guard counting frames no handler wants.
This commit is contained in:
Maurice
2026-08-10 21:34:30 +02:00
committed by Maurice
parent ba7d55f999
commit d981719941
18 changed files with 1188 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files
package-lock.json
Generated
+47 -4
View File
@@ -5400,6 +5400,25 @@
"url": "https://opencollective.com/express"
}
},
"node_modules/@nestjs/platform-ws": {
"version": "11.1.29",
"resolved": "https://registry.npmjs.org/@nestjs/platform-ws/-/platform-ws-11.1.29.tgz",
"integrity": "sha512-AYs3VZ5PxcDLiyGfvxsAS8DP9d7q+jxCKoWfMdl9RQLz69WAzSX2x+FgWNupTHt19zf8rmFXt2hvGd2tBUROAQ==",
"license": "MIT",
"dependencies": {
"tslib": "2.8.1",
"ws": "8.21.1"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/nest"
},
"peerDependencies": {
"@nestjs/common": "^11.0.0",
"@nestjs/websockets": "^11.0.0",
"rxjs": "^7.1.0"
}
},
"node_modules/@nestjs/swagger": {
"version": "11.4.5",
"resolved": "https://registry.npmjs.org/@nestjs/swagger/-/swagger-11.4.5.tgz",
@@ -5461,6 +5480,29 @@
}
}
},
"node_modules/@nestjs/websockets": {
"version": "11.1.29",
"resolved": "https://registry.npmjs.org/@nestjs/websockets/-/websockets-11.1.29.tgz",
"integrity": "sha512-Mg0sNK2k+PpamcJpDy9ZP24ZvM08xB6AZs061Kb6dhTesGxr6zrxlbZ7+40r5xwuxD/PN8LM2tSRKf5ojbFNDQ==",
"license": "MIT",
"dependencies": {
"iterare": "1.2.1",
"object-hash": "3.0.0",
"tslib": "2.8.1"
},
"peerDependencies": {
"@nestjs/common": "^11.0.0",
"@nestjs/core": "^11.0.0",
"@nestjs/platform-socket.io": "^11.0.0",
"reflect-metadata": "^0.1.12 || ^0.2.0",
"rxjs": "^7.1.0"
},
"peerDependenciesMeta": {
"@nestjs/platform-socket.io": {
"optional": true
}
}
},
"node_modules/@noble/hashes": {
"version": "1.8.0",
"resolved": "https://registry.npmjs.org/@noble/hashes/-/hashes-1.8.0.tgz",
@@ -15356,7 +15398,6 @@
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/object-hash/-/object-hash-3.0.0.tgz",
"integrity": "sha512-RSn9F68PjH9HqtltsSnqYC1XXoWe9Bju5+213R98cNGttag9q9yAOTzdbsqvIa7aNm5WffBZFpWYr2aWrklWAw==",
"dev": true,
"license": "MIT",
"engines": {
"node": ">= 6"
@@ -20939,9 +20980,9 @@
"license": "ISC"
},
"node_modules/ws": {
"version": "8.21.0",
"resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz",
"integrity": "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==",
"version": "8.21.1",
"resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz",
"integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==",
"license": "MIT",
"engines": {
"node": ">=10.0.0"
@@ -21156,7 +21197,9 @@
"@nestjs/config": "^4.0.4",
"@nestjs/core": "^11.1.24",
"@nestjs/platform-express": "^11.1.24",
"@nestjs/platform-ws": "^11.1.29",
"@nestjs/swagger": "^11.4.5",
"@nestjs/websockets": "^11.1.29",
"@simplewebauthn/server": "^13.1.2",
"@trek/nest-mcp": "*",
"@trek/shared": "*",
server/package.json
+2
View File
@@ -30,7 +30,9 @@
"@nestjs/config": "^4.0.4",
"@nestjs/core": "^11.1.24",
"@nestjs/platform-express": "^11.1.24",
"@nestjs/platform-ws": "^11.1.29",
"@nestjs/swagger": "^11.4.5",
"@nestjs/websockets": "^11.1.29",
"@simplewebauthn/server": "^13.1.2",
"@trek/nest-mcp": "*",
"@trek/shared": "*",
server/src/bootstrap.ts
+25
View File
@@ -1,3 +1,4 @@
import nodeHttp from 'node:http';
import { NestFactory } from '@nestjs/core';
import { ExpressAdapter, type NestExpressApplication } from '@nestjs/platform-express';
import type { INestApplication } from '@nestjs/common';
@@ -12,6 +13,7 @@ import { McpRegistryService } from '@trek/nest-mcp';
import { setMcpRegistry } from './mcp/registry-handoff';
import { validateBodyContracts } from './nest/common/validate-body-contracts';
import { validateRouteGuards } from './nest/common/validate-route-guards';
import { TrekWsAdapter } from './nest/realtime/trek-ws.adapter';
/**
* Builds the unified TREK NestJS application that serves the ENTIRE surface — the
@@ -43,12 +45,35 @@ import { validateRouteGuards } from './nest/common/validate-route-guards';
* SpaFallbackFilter (APP_FILTER in AppModule); the global error envelope is the
* TrekExceptionFilter (also APP_FILTER).
*/
let boundHttpServer: nodeHttp.Server | null = null;
/**
* The http server buildApp created and bound the ws gateway to.
*
* index.ts listens on it; the websocket suites connect to it. Anyone creating a
* second server around the same express instance would get a working REST API
* and a /ws that no client can reach.
*/
export function getHttpServer(): nodeHttp.Server {
if (!boundHttpServer) throw new Error('buildApp() has not run yet');
return boundHttpServer;
}
export async function buildApp(): Promise<INestApplication> {
// rawBody keeps the unparsed request bytes on req.rawBody so a plugin webhook
// route can verify a provider's HMAC signature over the exact payload (the
// parsed JSON alone can't be re-serialised byte-for-byte).
const app = await NestFactory.create(AppModule, new ExpressAdapter(), { rawBody: true });
const instance = app.getHttpAdapter().getInstance();
// The http server is created HERE, not by the caller after buildApp returns,
// and that ordering is the whole point: Nest binds gateways during app.init(),
// so the ws adapter has to exist before it and has to already know which
// server to attach to. Handing it the app instead would bind the ws server to
// Nest's own internal http server, which this process never listens on: the
// boot succeeds, the gateway logs as registered, every test passes, and no
// browser can connect. Callers take the server from getHttpServer() below.
boundHttpServer = nodeHttp.createServer(instance);
app.useWebSocketAdapter(new TrekWsAdapter(boundHttpServer));
// ConfigModule.forRoot's load factories already ran inside NestFactory.create,
// so the boot-stable snapshot is resolvable here, BEFORE app.init() — this is
// the one bridge that lets the pre-init Express layer consume the validated
server/src/index.ts
+5 -5
View File
@@ -8,7 +8,7 @@ import path from 'node:path';
import fs from 'node:fs';
import http from 'node:http';
import type { INestApplication } from '@nestjs/common';
import { buildApp } from './bootstrap';
import { buildApp, getHttpServer } from './bootstrap';
import { BackupService } from './nest/backup/backup.service';
import { PlacePhotoCacheService } from './nest/place-photos/place-photo-cache.service';
import { AirtrailSyncService } from './nest/integrations/airtrail-sync.service';
@@ -112,9 +112,6 @@ const onListen = () => {
scheduler.startTrekPhotoCacheCleanup();
scheduler.startPlacePhotoCacheCleanup();
scheduler.startAirTrailSync();
import('./websocket').then(({ setupWebSocket }) => {
setupWebSocket(server);
});
};
let server: http.Server;
@@ -127,7 +124,10 @@ async function bootstrap(): Promise<void> {
// (/mcp, /.well-known, OAuth SDK, SPA catch-all). buildApp() owns the composition
// order; it is shared with the integration-test harness so they can't drift.
nestApp = await buildApp();
server = http.createServer(nestApp.getHttpAdapter().getInstance());
// The server buildApp created and bound /ws to. Creating a second one here
// would serve the REST API fine and leave the gateway attached to a socket
// nobody listens on.
server = getHttpServer();
if (HOST) server.listen(PORT, HOST, onListen);
else server.listen(PORT, onListen);
}
server/src/nest/README.md
+38
View File
@@ -145,6 +145,44 @@ tools, which never pass through an HTTP guard. In the five domains piloted for
files of import churn, and `scripts/coverage-thresholds.mjs` buckets on one
path segment, so merging `photos/` into it would collapse two ratchet
entries into one and drop the stricter of the two.
- `realtime/` — the /ws transport, as a Nest gateway.
- `realtime.gateway.ts` owns the connection lifecycle and the handshake, with
`DatabaseService` and `EphemeralTokenService` injected instead of imported.
It replaced `setupWebSocket(server)`, which index.ts kicked off with a
dynamic import after listen().
- `ws-state.ts` holds the socket registry, and that is **module state on
purpose**. Seventeen `*.bridge.ts` files and `notifications.instance.ts`
build `new RealtimeService()` outside the container; rooms on a provider
instance would leave their broadcasts going to an empty map, with no error
and no log. Same reasoning as the geo throttle cursor.
- `trek-ws.adapter.ts` exists because the stock `WsAdapter` dispatches on
`{ event, data }` and every deployed client sends `{ type, tripId }`. It
also builds the ws server itself rather than delegating: the base's port-0
branch uses `noServer: true` and performs the upgrade from its own registry,
which never consults `verifyClient`, so delegating would silently drop the
origin check. The per-socket `error` listener is attached FIRST, because an
unhandled one kills the process (#1576), and the flood guard counts every
inbound frame before parsing, because malformed frames are the ones worth
limiting.
- The socket id stays a **monotonic integer**. The client echoes it as
`X-Socket-Id` and `broadcast` excludes the originator with
`Number(excludeSid)`; a uuid is `NaN`, `NaN === NaN` is false, and every
client would receive its own writes back. It shows up as drag-and-drop that
jumps, not as an error.
- `RealtimeGatewayModule` is imported by AppModule alone. RealtimeModule is
@Global, so a gateway there follows into every hand-assembled TestingModule,
and Nest's SocketModule then loads an adapter none of them set: it reaches
for `@nestjs/platform-socket.io` and calls `process.exit(1)` when it is
absent.
- `buildApp()` creates the http server and binds the adapter to it BEFORE
`app.init()`, and callers take it from `getHttpServer()`. Nest binds gateways
during init, so an adapter registered later binds nothing, and one given the
app instead of the server binds to Nest's internal server, which this process
never listens on. Either way the boot is green and no browser can connect.
- `src/websocket.ts` survives as a re-export. 115 test files `vi.mock` that
path to assert on broadcasts; repointing them all would be a large diff that
proves nothing.
- `geo/` — the ONE Nominatim client (#576). There used to be two: atlas
throttled to the published 1.1s and cached its answers, maps did neither, so
five interactive routes fired straight at the service while the atlas loops
server/src/nest/app.module.ts
+2 -1
View File
@@ -59,6 +59,7 @@ import { trekMcpAccessPolicy, trekMcpValidateAccess } from '../mcp/nest-mcp-poli
import { TrekExceptionFilter } from './common/trek-exception.filter';
import { SpaFallbackFilter } from './platform/spa-fallback.filter';
import { IdempotencyInterceptor } from './common/idempotency.interceptor';
import { RealtimeGatewayModule } from './realtime/realtime-gateway.module';
/**
* Root NestJS module for the incremental migration. Domain modules
@@ -66,7 +67,7 @@ import { IdempotencyInterceptor } from './common/idempotency.interceptor';
* migrated.
*/
@Module({
imports: [AppConfigModule, DatabaseModule, RealtimeModule, McpModule.forRoot({ accessPolicy: trekMcpAccessPolicy, validateAccess: trekMcpValidateAccess }), HealthModule, WeatherModule, HelpModule, AirportsModule, ConfigModule, SystemNoticesModule, GeoModule, MapsModule, PlaceEnrichmentModule, CategoriesModule, TagsModule, NotificationsModule, AtlasModule, VacayModule, PackingModule, TodoModule, BudgetModule, ReservationsModule, DaysModule, DayNotesModule, AccommodationsModule, AssignmentsModule, PlacesModule, TripsModule, CollabModule, FilesModule, PhotosModule, MemoriesModule, AirtrailModule, JourneyModule, CollectionsModule, ShareModule, TripInviteModule, TransitModule, FeedsModule, SettingsModule, BackupModule, AuthModule, OidcModule, OauthModule, AdminModule, AddonsModule, AuditModule, PermissionsModule, PluginsModule, BookingImportModule, ReservationImportModule, LlmParseModule],
imports: [AppConfigModule, DatabaseModule, RealtimeModule, RealtimeGatewayModule, McpModule.forRoot({ accessPolicy: trekMcpAccessPolicy, validateAccess: trekMcpValidateAccess }), HealthModule, WeatherModule, HelpModule, AirportsModule, ConfigModule, SystemNoticesModule, GeoModule, MapsModule, PlaceEnrichmentModule, CategoriesModule, TagsModule, NotificationsModule, AtlasModule, VacayModule, PackingModule, TodoModule, BudgetModule, ReservationsModule, DaysModule, DayNotesModule, AccommodationsModule, AssignmentsModule, PlacesModule, TripsModule, CollabModule, FilesModule, PhotosModule, MemoriesModule, AirtrailModule, JourneyModule, CollectionsModule, ShareModule, TripInviteModule, TransitModule, FeedsModule, SettingsModule, BackupModule, AuthModule, OidcModule, OauthModule, AdminModule, AddonsModule, AuditModule, PermissionsModule, PluginsModule, BookingImportModule, ReservationImportModule, LlmParseModule],
providers: [
// Default-deny: a route is authenticated unless it carries @Public() or
// @OptionalAuth(), or declares its own @UseGuards chain. Protection used to
server/src/nest/realtime/realtime-gateway.module.ts
+23
View File
@@ -0,0 +1,23 @@
import { Module } from '@nestjs/common';
import { RealtimeGateway } from './realtime.gateway';
import { EphemeralTokenModule } from '../auth/ephemeral-token.module';
/**
* The transport, kept out of RealtimeModule on purpose.
*
* RealtimeModule is @Global and every domain pulls it in for the broadcast
* facade. A gateway there would follow into every e2e TestingModule, and Nest's
* SocketModule loads a websocket adapter for any app that has one: with none
* set, it reaches for @nestjs/platform-socket.io and calls process.exit(1) when
* it is absent. Eighteen harnesses died that way.
*
* So this module is imported by AppModule alone. Test apps that assemble a few
* domain modules by hand get the facade without the transport, which is what
* they had before; the ones that boot the real buildApp() get both, with
* TrekWsAdapter already registered.
*/
@Module({
imports: [EphemeralTokenModule],
providers: [RealtimeGateway],
})
export class RealtimeGatewayModule {}
server/src/nest/realtime/realtime.gateway.ts
+161
View File
@@ -0,0 +1,161 @@
import { Injectable, OnModuleDestroy } from '@nestjs/common';
import {
ConnectedSocket,
MessageBody,
OnGatewayConnection,
OnGatewayDisconnect,
OnGatewayInit,
SubscribeMessage,
WebSocketGateway,
} from '@nestjs/websockets';
import type { IncomingMessage } from 'node:http';
import type { WebSocketServer } from 'ws';
import { DatabaseService } from '../database/database.service';
import { EphemeralTokenService } from '../auth/ephemeral-token.service';
import { User } from '../../types';
import {
joinRoom,
leaveAllRooms,
leaveRoom,
registerSocket,
userOf,
type TrekWebSocket,
} from './ws-state';
const HEARTBEAT_INTERVAL = 30_000;
/**
* The /ws transport, as a Nest gateway.
*
* It replaces the hand-rolled `setupWebSocket(server)` that index.ts kicked off
* with a dynamic import after listen(). What moved is ownership: the handshake
* reads its collaborators from the container instead of importing the db
* singleton and the token store, and the heartbeat is a lifecycle hook rather
* than an interval nobody stops.
*
* What did NOT move is the socket registry. It stays module-scoped in
* ws-state.ts because seventeen bridges construct `new RealtimeService()`
* outside the container and must see the same rooms; see the note there.
*
* The wire protocol is unchanged, down to the frame names. TrekWsAdapter maps
* `{ type }` onto @SubscribeMessage, because the stock adapter dispatches on
* `{ event, data }` and every deployed client speaks the former.
*/
@Injectable()
@WebSocketGateway({ path: '/ws' })
export class RealtimeGateway
implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect, OnModuleDestroy
{
private heartbeat: ReturnType<typeof setInterval> | null = null;
constructor(
private readonly db: DatabaseService,
private readonly tokens: EphemeralTokenService,
) {}
afterInit(server: WebSocketServer): void {
this.heartbeat = setInterval(() => {
server.clients.forEach((ws) => {
const tws = ws as TrekWebSocket;
if (tws.isAlive === false) return tws.terminate();
tws.isAlive = false;
tws.ping();
});
}, HEARTBEAT_INTERVAL);
this.heartbeat.unref?.();
}
onModuleDestroy(): void {
if (this.heartbeat) clearInterval(this.heartbeat);
this.heartbeat = null;
}
/**
* Authenticate the upgrade, then admit the socket.
*
* Every rejection closes with the code the client already handles: 4001 for
* anything about identity, 4403 for the MFA policy. The order matters and is
* unchanged — a missing token is refused before the store is touched, and the
* password-version gate runs before the MFA one.
*/
handleConnection(socket: TrekWebSocket, request: IncomingMessage): void {
const url = new URL(request.url ?? '/', 'http://localhost');
const token = url.searchParams.get('token');
if (!token) {
socket.close(4001, 'Authentication required');
return;
}
const consumed = this.tokens.consumeWithMeta(token, 'ws');
if (!consumed) {
socket.close(4001, 'Invalid or expired token');
return;
}
const row = this.db.get<User & { password_version?: number }>(
'SELECT id, username, email, role, mfa_enabled, password_version FROM users WHERE id = ?',
consumed.userId,
);
if (!row) {
socket.close(4001, 'User not found');
return;
}
// Session gate (defence-in-depth): reject a ws-token minted before a
// password change. Tokens carry the pv they were issued with; tokens minted
// without a pv (legacy) are treated as version 0, matching the JWT `pv`
// claim semantics in verifyJwtAndLoadUser.
const tokenPv = typeof consumed.pv === 'number' ? consumed.pv : 0;
const currentPv = typeof row.password_version === 'number' ? row.password_version : 0;
if (tokenPv !== currentPv) {
socket.close(4001, 'Invalid or expired token');
return;
}
// Don't leak password_version beyond the handshake.
const { password_version: _pv, ...user } = row;
const requireMfa =
this.db.get<{ value: string }>("SELECT value FROM app_settings WHERE key = 'require_mfa'")?.value === 'true';
const mfaOk = user.mfa_enabled === 1 || user.mfa_enabled === true;
if (requireMfa && !mfaOk) {
socket.close(4403, 'MFA required');
return;
}
socket.isAlive = true;
const sid = registerSocket(socket, user as User);
socket.send(JSON.stringify({ type: 'welcome', socketId: sid }));
socket.on('pong', () => { socket.isAlive = true; });
}
handleDisconnect(socket: TrekWebSocket): void {
leaveAllRooms(socket);
}
@SubscribeMessage('join')
handleJoin(
@MessageBody() message: { tripId?: number | string },
@ConnectedSocket() socket: TrekWebSocket,
): { type: string; tripId?: number; message?: string } | undefined {
const user = userOf(socket);
if (!user || !message?.tripId) return undefined;
const tripId = Number(message.tripId);
if (!this.db.canAccessTrip(tripId, user.id)) {
return { type: 'error', message: 'Access denied' };
}
joinRoom(socket, tripId);
return { type: 'joined', tripId };
}
@SubscribeMessage('leave')
handleLeave(
@MessageBody() message: { tripId?: number | string },
@ConnectedSocket() socket: TrekWebSocket,
): { type: string; tripId?: number; message?: string } | undefined {
if (!message?.tripId) return undefined;
const tripId = Number(message.tripId);
leaveRoom(socket, tripId);
return { type: 'left', tripId };
}
}
server/src/nest/realtime/realtime.module.ts
+8 -4
View File
@@ -3,10 +3,14 @@ import { RealtimeService } from './realtime.service';
/**
* Global so every migrated module can inject RealtimeService without
* re-importing (same rationale as DatabaseModule). The service is
* stateless and dependency-free — the ws server itself is still wired
* in index.ts via setupWebSocket(); this module only provides the
* injectable broadcast seam.
* re-importing (same rationale as DatabaseModule). The service stays
* stateless and dependency-free so the out-of-container bridges can build it
* with a bare `new`.
*
* The transport itself lives in RealtimeGatewayModule, which AppModule imports
* directly. It cannot live here: this module is @Global, so a gateway would
* follow into every hand-assembled TestingModule and make Nest look for a
* websocket adapter none of them set.
*/
@Global()
@Module({
server/src/nest/realtime/trek-ws.adapter.ts
+161
View File
@@ -0,0 +1,161 @@
import { WsAdapter } from '@nestjs/platform-ws';
import type { INestApplicationContext } from '@nestjs/common';
import type { Server as HttpServer } from 'node:http';
import type { MessageMappingProperties } from '@nestjs/websockets';
import { WebSocketServer } from 'ws';
import type { Observable } from 'rxjs';
import { readEnv } from '../../app-config';
import { setServer, type TrekWebSocket } from './ws-state';
// Per-connection message rate limiting. It lives in the adapter, not in the
// gateway's handlers, because the original counted EVERY inbound frame before
// parsing it: malformed JSON and unknown types count too. Counting only the
// frames that reach a @SubscribeMessage handler would leave a client free to
// flood the process with garbage.
const WS_MSG_LIMIT = 30; // max messages
const WS_MSG_WINDOW = 10_000; // per 10 seconds
const rates = new WeakMap<TrekWebSocket, { count: number; windowStart: number }>();
function withinRate(socket: TrekWebSocket): boolean {
let rate = rates.get(socket);
if (!rate) {
rate = { count: 0, windowStart: Date.now() };
rates.set(socket, rate);
}
const now = Date.now();
if (now - rate.windowStart > WS_MSG_WINDOW) {
rate.count = 1;
rate.windowStart = now;
return true;
}
rate.count++;
return rate.count <= WS_MSG_LIMIT;
}
/**
* TREK's wire protocol, on Nest's gateway plumbing.
*
* The stock WsAdapter dispatches on `{ event, data }`. TREK's client has always
* sent `{ type: 'join', tripId }` and expects `{ type: 'joined' }` back, and
* changing that would break every deployed client and every offline queue that
* has a frame parked in it. So the adapter translates: `type` names the handler,
* and the whole message is the payload.
*
* It also owns two things the stock adapter has no concept of, both of which
* predate this migration and neither of which may be dropped:
*
* - The origin check. `verifyClient` rejects a cross-origin upgrade with 403
* before a socket exists, which is cheaper and stricter than checking after.
* - The per-socket `error` listener, attached FIRST. `ws` surfaces protocol
* violations (malformed frames, reserved close codes such as 1006) as an
* `error` event on the socket; unhandled, Node rethrows it and the process
* dies. That is #1576, and it only reproduces against a hostile client, so
* nothing in CI would catch its removal.
*/
export class TrekWsAdapter extends WsAdapter {
private readonly httpServerRef: HttpServer;
/**
* Takes the http.Server directly, not the app.
*
* The base adapter accepts either, but only the server form guarantees the ws
* server lands on the socket this process actually listens on. Given the app,
* it would reach for Nest's own internal server, which buildApp never uses.
*/
constructor(httpServer: HttpServer) {
super(httpServer as unknown as INestApplicationContext);
this.httpServerRef = httpServer;
}
/**
* Builds the ws server directly rather than delegating to the base.
*
* The base's port-0 branch creates it with `noServer: true` and performs the
* upgrade itself from its own registry. That path never consults
* `verifyClient`, because ws only calls it when ws owns the upgrade — so
* delegating would silently drop the origin check and answer 101 to any
* origin. Attaching to the http server keeps ws in charge of the handshake,
* which is where the 403 has always come from.
*/
create(_port: number, options?: Record<string, unknown>): unknown {
const allowedOrigins = readEnv().http.wsOrigins;
const server = new WebSocketServer({
server: this.httpServerRef,
path: (options?.path as string) ?? '/ws',
maxPayload: 64 * 1024, // 64 KB max message size
...(allowedOrigins
? {
verifyClient: (
{ origin }: { origin: string },
cb: (ok: boolean, code?: number, msg?: string) => void,
) => {
if (!origin || allowedOrigins.includes(origin)) cb(true);
else cb(false, 403, 'Origin not allowed');
},
}
: {}),
});
// A server-level error must not reach the process. The base binds one on the
// instances it creates; this one is ours, so it needs its own.
server.on('error', () => {});
// The fan-out primitives reach the live server through here, and so do the
// out-of-container bridges.
setServer(server);
return server;
}
bindClientConnect(server: WebSocketServer, callback: (...args: unknown[]) => void): void {
server.on('connection', (socket: TrekWebSocket, request: unknown) => {
// Must stay above anything that can close early: a socket that never gets
// this listener can still crash the process while it finishes closing.
socket.on('error', () => socket.terminate());
// The request rides along: the handshake reads the ws token off its query
// string, so dropping it here would leave handleConnection with nothing
// to authenticate.
callback(socket, request);
});
}
bindMessageHandlers(
socket: TrekWebSocket,
handlers: MessageMappingProperties[],
transform: (data: unknown) => Observable<unknown>,
): void {
socket.on('message', (buffer: Buffer) => {
if (!withinRate(socket)) {
if (socket.readyState === 1) {
socket.send(JSON.stringify({ type: 'error', message: 'Rate limit exceeded' }));
}
return;
}
let message: { type?: unknown };
try {
message = JSON.parse(buffer.toString());
} catch {
return; // Malformed JSON, ignored exactly as before.
}
if (!message || typeof message !== 'object' || typeof message.type !== 'string') return;
const handler = handlers.find((h) => h.message === message.type);
if (!handler) return; // An unknown type is ignored, not an error frame.
// The whole message is the payload: TREK's frames are flat, so `tripId`
// sits beside `type` rather than under a `data` key.
transform(handler.callback(message, socket)).subscribe({
next: (response) => {
if (response !== undefined && socket.readyState === 1) {
socket.send(JSON.stringify(response));
}
},
});
});
}
async close(server: WebSocketServer): Promise<void> {
setServer(null);
await super.close(server);
}
}
server/src/nest/realtime/ws-state.ts
+143
View File
@@ -0,0 +1,143 @@
import { WebSocketServer, WebSocket } from 'ws';
import { emitPluginEvent, pluginEventMeta } from '../../plugin-event-sink';
import { User } from '../../types';
/**
* The socket registry: rooms, per-socket identity, and the three fan-out
* primitives.
*
* MODULE state, not provider state, and that is load-bearing. Seventeen
* `*.bridge.ts` files and notifications.instance.ts build `new
* RealtimeService()` by hand, outside the container, because they run outside
* it. If the rooms lived on a provider instance, those broadcasts would go to
* an empty map: no error, no log, just a client that stops updating. Same
* reasoning as the geo throttle cursor and the permissions cache.
*
* The gateway next door owns the connection lifecycle and the handshake and
* writes into here. This file deliberately knows nothing about auth.
*/
export interface TrekWebSocket extends WebSocket {
isAlive: boolean;
}
const rooms = new Map<number, Set<TrekWebSocket>>();
const socketRooms = new WeakMap<TrekWebSocket, Set<number>>();
const socketUser = new WeakMap<TrekWebSocket, User>();
const socketId = new WeakMap<TrekWebSocket, number>();
/**
* A monotonic integer, NOT a uuid.
*
* The client echoes it back as the X-Socket-Id header and `broadcast` excludes
* the originator with `Number(excludeSid)`. `Number(<uuid>)` is NaN, and
* `NaN === NaN` is false, so every client would receive its own writes back.
* Nothing throws; it shows up as drag-and-drop that jumps back under your
* cursor.
*/
let nextSocketId = 1;
let wss: WebSocketServer | null = null;
export function setServer(server: WebSocketServer | null): void {
wss = server;
}
export function getServer(): WebSocketServer | null {
return wss;
}
export function registerSocket(ws: TrekWebSocket, user: User): number {
const sid = nextSocketId++;
socketId.set(ws, sid);
socketUser.set(ws, user);
socketRooms.set(ws, new Set());
return sid;
}
export function userOf(ws: TrekWebSocket): User | undefined {
return socketUser.get(ws);
}
export function joinRoom(ws: TrekWebSocket, tripId: number): void {
if (!rooms.has(tripId)) rooms.set(tripId, new Set());
rooms.get(tripId)!.add(ws);
socketRooms.get(ws)?.add(tripId);
}
export function leaveRoom(ws: TrekWebSocket, tripId: number): void {
const room = rooms.get(tripId);
if (room) {
room.delete(ws);
if (room.size === 0) rooms.delete(tripId);
}
socketRooms.get(ws)?.delete(tripId);
}
export function leaveAllRooms(ws: TrekWebSocket): void {
const mine = socketRooms.get(ws);
if (!mine) return;
for (const tripId of mine) leaveRoom(ws, tripId);
}
/**
* Broadcast an event to all sockets in a trip room, optionally excluding a
* socket.
*
* When `onlyUserId` is given the event is delivered only to that user's sockets
* in the room — used to keep private packing items (#858) off other members'
* screens while still syncing the owner's own tabs.
*/
export function broadcast(
tripId: number | string,
eventType: string,
payload: Record<string, unknown>,
excludeSid?: number | string,
onlyUserId?: number,
): void {
tripId = Number(tripId);
// Announce every CORE trip event (name only, never the payload) to subscribed
// plugins — before the room check so it fires even with no connected viewers
// and with no ws server at all, and skipping plugin:* re-broadcasts so a
// plugin's own events can't loop back.
if (!eventType.startsWith('plugin:')) emitPluginEvent(tripId, eventType, pluginEventMeta(eventType, payload));
const room = rooms.get(tripId);
if (!room || room.size === 0) return;
const excludeNum = excludeSid ? Number(excludeSid) : null;
for (const ws of room) {
if (ws.readyState !== 1) continue; // WebSocket.OPEN === 1
if (excludeNum && socketId.get(ws) === excludeNum) continue;
if (onlyUserId != null && socketUser.get(ws)?.id !== onlyUserId) continue;
ws.send(JSON.stringify({ type: eventType, tripId, ...payload }));
}
}
/** Send a message to all sockets belonging to a specific user (e.g. trip invitations). */
export function broadcastToUser(
userId: number,
payload: Record<string, unknown>,
excludeSid?: number | string,
): void {
if (!wss) return;
const excludeNum = excludeSid ? Number(excludeSid) : null;
for (const ws of wss.clients) {
const tws = ws as TrekWebSocket;
if (tws.readyState !== 1) continue;
if (excludeNum && socketId.get(tws) === excludeNum) continue;
if (socketUser.get(tws)?.id === userId) tws.send(JSON.stringify(payload));
}
}
export function getOnlineUserIds(): Set<number> {
const ids = new Set<number>();
if (!wss) return ids;
for (const ws of wss.clients) {
const tws = ws as TrekWebSocket;
if (tws.readyState !== 1) continue;
const user = socketUser.get(tws);
if (user) ids.add(user.id);
}
return ids;
}
server/src/websocket.ts
+12 -245
View File
@@ -1,247 +1,14 @@
import { WebSocketServer, WebSocket } from 'ws';
import { readEnv } from './app-config';
import { db, canAccessTrip } from './db/database';
import { consumeEphemeralTokenWithMeta } from './nest/auth/ephemeral-tokens';
import { emitPluginEvent, pluginEventMeta } from './plugin-event-sink';
import { User } from './types';
import http from 'node:http';
interface NomadWebSocket extends WebSocket {
isAlive: boolean;
}
// Room management: tripId -> Set<WebSocket>
const rooms = new Map<number, Set<NomadWebSocket>>();
// Track which rooms each socket is in
const socketRooms = new WeakMap<NomadWebSocket, Set<number>>();
// Track user info per socket
const socketUser = new WeakMap<NomadWebSocket, User>();
// Track unique socket ID
const socketId = new WeakMap<NomadWebSocket, number>();
let nextSocketId = 1;
let wss: WebSocketServer | null = null;
// Per-connection message rate limiting
const WS_MSG_LIMIT = 30; // max messages
const WS_MSG_WINDOW = 10_000; // per 10 seconds
const socketMsgCounts = new WeakMap<NomadWebSocket, { count: number; windowStart: number }>();
/** Attaches a WebSocket server with JWT auth, room-based trip channels, and heartbeat keep-alive. */
function setupWebSocket(server: http.Server): void {
const allowedOrigins = readEnv().http.wsOrigins;
wss = new WebSocketServer({
server,
path: '/ws',
maxPayload: 64 * 1024, // 64 KB max message size
verifyClient: allowedOrigins
? ({ origin }, cb) => {
if (!origin || allowedOrigins.includes(origin)) cb(true);
else cb(false, 403, 'Origin not allowed');
}
: undefined,
});
const HEARTBEAT_INTERVAL = 30000; // 30 seconds
const heartbeat = setInterval(() => {
wss.clients.forEach((ws) => {
const nws = ws as NomadWebSocket;
if (nws.isAlive === false) return nws.terminate();
nws.isAlive = false;
nws.ping();
});
}, HEARTBEAT_INTERVAL);
wss.on('close', () => clearInterval(heartbeat));
wss.on('connection', (ws: WebSocket, req: http.IncomingMessage) => {
const nws = ws as NomadWebSocket;
// ws emits protocol violations (malformed frames, reserved close codes such as 1006)
// as an 'error' event on the socket; unhandled, Node rethrows it and the process dies.
// Must stay above the auth guards below: they close and return early, and a socket that
// never gets this listener can still crash the server before it finishes closing.
nws.on('error', () => nws.terminate());
// Extract token from query param
const url = new URL(req.url, 'http://localhost');
const token = url.searchParams.get('token');
if (!token) {
nws.close(4001, 'Authentication required');
return;
}
const consumed = consumeEphemeralTokenWithMeta(token, 'ws');
if (!consumed) {
nws.close(4001, 'Invalid or expired token');
return;
}
const { userId } = consumed;
let row: (User & { password_version?: number }) | undefined;
row = db.prepare(
'SELECT id, username, email, role, mfa_enabled, password_version FROM users WHERE id = ?'
).get(userId) as (User & { password_version?: number }) | undefined;
if (!row) {
nws.close(4001, 'User not found');
return;
}
// Session gate (defence-in-depth): reject a ws-token minted before a
// password change. Tokens carry the pv they were issued with; tokens
// minted without a pv (legacy) are treated as version 0, matching the
// JWT `pv` claim semantics in verifyJwtAndLoadUser.
const tokenPv = typeof consumed.pv === 'number' ? consumed.pv : 0;
const currentPv = typeof row.password_version === 'number' ? row.password_version : 0;
if (tokenPv !== currentPv) {
nws.close(4001, 'Invalid or expired token');
return;
}
// Don't leak password_version beyond the handshake.
const { password_version: _pv, ...user } = row;
const requireMfa = (db.prepare("SELECT value FROM app_settings WHERE key = 'require_mfa'").get() as { value: string } | undefined)?.value === 'true';
const mfaOk = user.mfa_enabled === 1 || user.mfa_enabled === true;
if (requireMfa && !mfaOk) {
nws.close(4403, 'MFA required');
return;
}
nws.isAlive = true;
const sid = nextSocketId++;
socketId.set(nws, sid);
socketUser.set(nws, user);
socketRooms.set(nws, new Set());
nws.send(JSON.stringify({ type: 'welcome', socketId: sid }));
nws.on('pong', () => { nws.isAlive = true; });
socketMsgCounts.set(nws, { count: 0, windowStart: Date.now() });
nws.on('message', (data) => {
// Rate limiting
const rate = socketMsgCounts.get(nws);
const now = Date.now();
if (now - rate.windowStart > WS_MSG_WINDOW) {
rate.count = 1;
rate.windowStart = now;
} else {
rate.count++;
if (rate.count > WS_MSG_LIMIT) {
nws.send(JSON.stringify({ type: 'error', message: 'Rate limit exceeded' }));
return;
}
}
let msg: { type: string; tripId?: number | string };
try {
msg = JSON.parse(data.toString());
} catch {
return; // Malformed JSON, ignore
}
// Basic validation
if (!msg || typeof msg !== 'object' || typeof msg.type !== 'string') return;
if (msg.type === 'join' && msg.tripId) {
const tripId = Number(msg.tripId);
// Verify the user has access to this trip
if (!canAccessTrip(tripId, user.id)) {
nws.send(JSON.stringify({ type: 'error', message: 'Access denied' }));
return;
}
// Add to room
if (!rooms.has(tripId)) rooms.set(tripId, new Set());
rooms.get(tripId).add(nws);
socketRooms.get(nws).add(tripId);
nws.send(JSON.stringify({ type: 'joined', tripId }));
}
if (msg.type === 'leave' && msg.tripId) {
const tripId = Number(msg.tripId);
leaveRoom(nws, tripId);
nws.send(JSON.stringify({ type: 'left', tripId }));
}
});
nws.on('close', () => {
// Clean up all rooms this socket was in
const myRooms = socketRooms.get(nws);
if (myRooms) {
for (const tripId of myRooms) {
leaveRoom(nws, tripId);
}
}
});
});
console.log('WebSocket server attached at /ws');
}
function leaveRoom(ws: NomadWebSocket, tripId: number): void {
const room = rooms.get(tripId);
if (room) {
room.delete(ws);
if (room.size === 0) rooms.delete(tripId);
}
const myRooms = socketRooms.get(ws);
if (myRooms) myRooms.delete(tripId);
}
/**
* Broadcast an event to all sockets in a trip room, optionally excluding a socket.
* When `onlyUserId` is given the event is delivered only to that user's sockets in
* the room — used to keep private packing items (#858) off other members' screens
* while still syncing the owner's own tabs.
* The transport's public surface, for code outside the Nest container.
*
* The implementation moved to nest/realtime: the connection lifecycle and the
* handshake are RealtimeGateway, the wire protocol is TrekWsAdapter, and the
* socket registry is ws-state.ts. This file stays because it is the import path
* the rest of the tree knows, including 115 test files that `vi.mock` it to
* assert on broadcasts. Repointing them all would be a large diff that proves
* nothing, and this re-export keeps every one of those mocks working.
*
* `setupWebSocket` is gone: the server is created by the adapter now, when Nest
* initialises the gateway, rather than by index.ts after listen().
*/
function broadcast(tripId: number | string, eventType: string, payload: Record<string, unknown>, excludeSid?: number | string, onlyUserId?: number): void {
tripId = Number(tripId);
// Announce every CORE trip event (name only, never the payload) to subscribed
// plugins — before the room check so it fires even with no connected viewers, and
// skipping plugin:* re-broadcasts so a plugin's own events can't loop back.
if (!eventType.startsWith('plugin:')) emitPluginEvent(tripId, eventType, pluginEventMeta(eventType, payload));
const room = rooms.get(tripId);
if (!room || room.size === 0) return;
const excludeNum = excludeSid ? Number(excludeSid) : null;
for (const ws of room) {
if (ws.readyState !== 1) continue; // WebSocket.OPEN === 1
// Exclude the specific socket that triggered the change
if (excludeNum && socketId.get(ws) === excludeNum) continue;
if (onlyUserId != null && socketUser.get(ws)?.id !== onlyUserId) continue;
ws.send(JSON.stringify({ type: eventType, tripId, ...payload }));
}
}
/** Send a message to all sockets belonging to a specific user (e.g., for trip invitations). */
function broadcastToUser(userId: number, payload: Record<string, unknown>, excludeSid?: number | string): void {
if (!wss) return;
const excludeNum = excludeSid ? Number(excludeSid) : null;
for (const ws of wss.clients) {
const nws = ws as NomadWebSocket;
if (nws.readyState !== 1) continue;
if (excludeNum && socketId.get(nws) === excludeNum) continue;
const user = socketUser.get(nws);
if (user?.id === userId) {
nws.send(JSON.stringify(payload));
}
}
}
function getOnlineUserIds(): Set<number> {
const ids = new Set<number>();
if (!wss) return ids;
for (const ws of wss.clients) {
const nws = ws as NomadWebSocket;
if (nws.readyState !== 1) continue;
const user = socketUser.get(nws);
if (user) ids.add(user.id);
}
return ids;
}
export { setupWebSocket, broadcast, broadcastToUser, getOnlineUserIds };
export { broadcast, broadcastToUser, getOnlineUserIds } from './nest/realtime/ws-state';
server/tests/helpers/ws-client.ts
+2 -2
View File
@@ -3,7 +3,7 @@
*
* Usage:
* import http from 'http';
* import { setupWebSocket } from '../../src/websocket';
* import { getHttpServer } from '../../src/bootstrap';
* import { WsTestClient, getWsToken } from '../helpers/ws-client';
*
* let server: http.Server;
@@ -12,7 +12,7 @@
* beforeAll(async () => {
* const app = createApp();
* server = http.createServer(app);
* setupWebSocket(server);
* const server = getHttpServer();
* await new Promise<void>(res => server.listen(0, res));
* });
*
server/tests/integration/journey.test.ts
-1
View File
@@ -52,7 +52,6 @@ vi.mock('../../src/config', () => ({
vi.mock('../../src/websocket', () => ({
broadcast: vi.fn(),
broadcastToUser: vi.fn(),
setupWebSocket: vi.fn(),
getOnlineUserIds: vi.fn(() => []),
}));
vi.mock('../../src/services/memories/immichService', () => ({
server/tests/unit/nest/realtime.gateway.test.ts
+288
View File
@@ -0,0 +1,288 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
/**
* The gateway's handshake, and the socket registry it writes into.
*
* The transport suites in tests/websocket drive real sockets end to end; these
* cases go at the pieces directly, so the ways this can fail SILENTLY are
* pinned by name rather than by whether a browser happened to work.
*/
vi.mock('../../../src/plugin-event-sink', () => ({
emitPluginEvent: vi.fn(),
pluginEventMeta: vi.fn(() => ({})),
}));
import { RealtimeGateway } from '../../../src/nest/realtime/realtime.gateway';
import {
broadcast,
broadcastToUser,
getOnlineUserIds,
joinRoom,
registerSocket,
setServer,
type TrekWebSocket,
} from '../../../src/nest/realtime/ws-state';
import { emitPluginEvent } from '../../../src/plugin-event-sink';
import type { DatabaseService } from '../../../src/nest/database/database.service';
import type { EphemeralTokenService } from '../../../src/nest/auth/ephemeral-token.service';
import type { User } from '../../../src/types';
interface FakeSocket extends TrekWebSocket {
sent: string[];
closedWith: [number, string] | null;
}
function socket(): FakeSocket {
const sent: string[] = [];
const s = {
readyState: 1,
isAlive: false,
sent,
closedWith: null as [number, string] | null,
send: (raw: string) => { sent.push(raw); },
close: (code: number, reason: string) => { s.closedWith = [code, reason]; },
on: vi.fn(),
terminate: vi.fn(),
ping: vi.fn(),
};
return s as unknown as FakeSocket;
}
const rows = new Map<string, unknown>();
const db = {
get: (sql: string) => (sql.includes('app_settings') ? rows.get('mfa') : rows.get('user')),
canAccessTrip: (tripId: number) => tripId === 7,
} as unknown as DatabaseService;
const consumeWithMeta = vi.fn();
const tokens = { consumeWithMeta } as unknown as EphemeralTokenService;
function connect(url: string) {
const gw = new RealtimeGateway(db, tokens);
const ws = socket();
gw.handleConnection(ws, { url } as never);
return { gw, ws };
}
beforeEach(() => {
vi.clearAllMocks();
rows.clear();
rows.set('user', { id: 3, username: 'm', email: 'm@x.test', role: 'user', mfa_enabled: 0, password_version: 2 });
consumeWithMeta.mockReturnValue({ userId: 3, pv: 2 });
});
describe('RealtimeGateway handshake', () => {
it('WSGW-001: refuses a connect with no token before touching the store', () => {
const { ws } = connect('/ws');
expect(ws.closedWith).toEqual([4001, 'Authentication required']);
expect(consumeWithMeta).not.toHaveBeenCalled();
});
it('WSGW-002: refuses an unknown or spent token', () => {
consumeWithMeta.mockReturnValue(null);
expect(connect('/ws?token=x').ws.closedWith).toEqual([4001, 'Invalid or expired token']);
});
it('WSGW-003: rejects a token minted before a password change', () => {
// The pv gate. Same close reason as an unknown token on purpose: a client
// must not be able to tell a stale token from a forged one.
consumeWithMeta.mockReturnValue({ userId: 3, pv: 1 });
expect(connect('/ws?token=x').ws.closedWith).toEqual([4001, 'Invalid or expired token']);
});
it('WSGW-004: treats a token minted without a pv as version 0', () => {
consumeWithMeta.mockReturnValue({ userId: 3 });
rows.set('user', { id: 3, email: 'm@x.test', role: 'user', mfa_enabled: 0, password_version: 0 });
expect(connect('/ws?token=x').ws.closedWith).toBeNull();
});
it('WSGW-005: enforces the MFA policy with its own close code', () => {
rows.set('mfa', { value: 'true' });
expect(connect('/ws?token=x').ws.closedWith).toEqual([4403, 'MFA required']);
});
it('WSGW-006: admits an MFA-enabled user while the policy is on', () => {
rows.set('mfa', { value: 'true' });
rows.set('user', { id: 3, email: 'm@x.test', role: 'user', mfa_enabled: 1, password_version: 2 });
expect(connect('/ws?token=x').ws.closedWith).toBeNull();
});
it('WSGW-007: the welcome frame carries a NUMERIC socket id', () => {
// Load-bearing: the client echoes this back as X-Socket-Id and broadcast
// excludes the originator with Number(excludeSid). A uuid would be NaN,
// NaN === NaN is false, and every client would receive its own writes back.
// Nothing throws; it shows up as drag-and-drop that jumps under the cursor.
const { ws } = connect('/ws?token=x');
const welcome = JSON.parse(ws.sent[0]) as { type: string; socketId: unknown };
expect(welcome.type).toBe('welcome');
expect(Number.isInteger(welcome.socketId)).toBe(true);
});
it('WSGW-007b: survives an upgrade request with no url, rather than throwing at it', () => {
const gw = new RealtimeGateway(db, tokens);
const ws = socket();
expect(() => gw.handleConnection(ws, {} as never)).not.toThrow();
expect(ws.closedWith).toEqual([4001, 'Authentication required']);
});
it('WSGW-007c: a user row with no password_version reads as version 0', () => {
// Legacy rows predate the column. Treating a missing value as 0 is what
// makes a legacy token match a legacy row instead of being refused forever.
consumeWithMeta.mockReturnValue({ userId: 3, pv: 0 });
rows.set('user', { id: 3, email: 'm@x.test', role: 'user', mfa_enabled: 0 });
expect(connect('/ws?token=x').ws.closedWith).toBeNull();
});
it('WSGW-008: never leaks password_version past the handshake', () => {
const { gw, ws } = connect('/ws?token=x');
gw.handleJoin({ tripId: 7 }, ws);
broadcast(7, 'trip:updated' as never, {} as never);
expect(ws.sent.join('')).not.toContain('password_version');
});
});
describe('RealtimeGateway rooms', () => {
it('WSGW-010: join refuses a trip the user cannot reach', () => {
const { gw, ws } = connect('/ws?token=x');
expect(gw.handleJoin({ tripId: 99 }, ws)).toEqual({ type: 'error', message: 'Access denied' });
});
it('WSGW-011: join and leave answer flat, keyed by type', () => {
const { gw, ws } = connect('/ws?token=x');
expect(gw.handleJoin({ tripId: 7 }, ws)).toEqual({ type: 'joined', tripId: 7 });
expect(gw.handleLeave({ tripId: 7 }, ws)).toEqual({ type: 'left', tripId: 7 });
});
it('WSGW-012: a frame with no tripId is ignored rather than answered', () => {
const { gw, ws } = connect('/ws?token=x');
expect(gw.handleJoin({}, ws)).toBeUndefined();
expect(gw.handleLeave({}, ws)).toBeUndefined();
});
});
describe('RealtimeGateway heartbeat', () => {
it('WSGW-020: pings a live socket and terminates one that missed the last ping', () => {
vi.useFakeTimers();
try {
const alive = socket();
alive.isAlive = true;
const stale = socket();
stale.isAlive = false;
const gw = new RealtimeGateway(db, tokens);
gw.afterInit({ clients: new Set([alive, stale]) } as never);
vi.advanceTimersByTime(30_000);
// A socket that answered the last ping is asked again and marked pending.
expect(alive.ping).toHaveBeenCalledTimes(1);
expect(alive.isAlive).toBe(false);
// One that never answered is dropped rather than pinged forever.
expect(stale.terminate).toHaveBeenCalledTimes(1);
expect(stale.ping).not.toHaveBeenCalled();
} finally {
vi.useRealTimers();
}
});
it('WSGW-021: the interval stops with the module, so a test process can exit', () => {
vi.useFakeTimers();
try {
const ws = socket();
ws.isAlive = true;
const gw = new RealtimeGateway(db, tokens);
gw.afterInit({ clients: new Set([ws]) } as never);
gw.onModuleDestroy();
vi.advanceTimersByTime(90_000);
expect(ws.ping).not.toHaveBeenCalled();
// Idempotent: shutdown can run twice without clearing a stale handle.
expect(() => gw.onModuleDestroy()).not.toThrow();
} finally {
vi.useRealTimers();
}
});
it('WSGW-022: a pong marks the socket live again', () => {
const { ws } = connect('/ws?token=x');
const pong = (ws.on as unknown as { mock: { calls: [string, () => void][] } }).mock.calls
.find(([event]) => event === 'pong');
expect(pong).toBeDefined();
ws.isAlive = false;
pong![1]();
expect(ws.isAlive).toBe(true);
});
});
describe('ws-state fan-out', () => {
it('WSST-001: broadcast excludes the originating socket by its numeric id', () => {
const a = socket();
const b = socket();
const sidA = registerSocket(a, { id: 1 } as User);
registerSocket(b, { id: 2 } as User);
joinRoom(a, 42);
joinRoom(b, 42);
broadcast(42, 'place:created' as never, { placeId: 9 } as never, sidA);
expect(a.sent).toHaveLength(0);
expect(JSON.parse(b.sent[0])).toMatchObject({ type: 'place:created', tripId: 42, placeId: 9 });
});
it('WSST-002: onlyUserId narrows delivery to one member', () => {
const mine = socket();
const theirs = socket();
registerSocket(mine, { id: 1 } as User);
registerSocket(theirs, { id: 2 } as User);
joinRoom(mine, 43);
joinRoom(theirs, 43);
broadcast(43, 'packing:updated' as never, {} as never, undefined, 1);
expect(mine.sent).toHaveLength(1);
expect(theirs.sent).toHaveLength(0);
});
it('WSST-003: announces core events to plugins BEFORE the room check', () => {
// No room, no viewers, and with a plugin subscribed the event still has to
// fire. Ordering it after the early return would silence plugin events on
// every trip nobody happens to be looking at.
broadcast(9999, 'day:updated' as never, {} as never);
expect(emitPluginEvent).toHaveBeenCalledWith(9999, 'day:updated', expect.anything());
});
it('WSST-004: does not re-announce a plugin broadcast, which would loop', () => {
broadcast(9998, 'plugin:acme:ping' as never, {} as never);
expect(emitPluginEvent).not.toHaveBeenCalled();
});
it('WSST-004b: a closing socket is skipped by both fan-outs', () => {
const open = socket();
const closing = socket();
(closing as { readyState: number }).readyState = 2; // CLOSING
registerSocket(open, { id: 1 } as User);
registerSocket(closing, { id: 2 } as User);
joinRoom(open, 44);
joinRoom(closing, 44);
setServer({ clients: new Set([open, closing]) } as never);
broadcast(44, 'trip:updated' as never, {} as never);
broadcastToUser(2, { type: 'trip:invite' });
expect(open.sent).toHaveLength(1);
expect(closing.sent).toHaveLength(0);
// And it is not counted as online, which drives the admin presence dots.
expect(getOnlineUserIds()).toEqual(new Set([1]));
});
it('WSST-004c: a socket with no registered user is not reported online', () => {
const stray = socket();
setServer({ clients: new Set([stray]) } as never);
expect(getOnlineUserIds()).toEqual(new Set());
});
it('WSST-005: broadcastToUser and getOnlineUserIds answer empty with no server', () => {
setServer(null);
expect(() => broadcastToUser(1, { type: 'x' })).not.toThrow();
expect(getOnlineUserIds()).toEqual(new Set());
});
});
server/tests/unit/nest/trek-ws-adapter.test.ts
+263
View File
@@ -0,0 +1,263 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
/**
* TrekWsAdapter: the wire protocol and the per-socket flood guard.
*
* Both are things the stock WsAdapter does differently, and both fail quietly
* if this class regresses. A protocol slip means `join` is never handled, the
* room stays empty, `broadcast` returns at its size check and every client
* simply stops updating. A missing flood guard is only visible under a hostile
* client.
*/
// The origin allowlist is env-driven and empty in the test environment, so the
// branch that builds verifyClient would never run. Driving readEnv lets both
// sides of it be asserted rather than assumed.
const { wsOrigins } = vi.hoisted(() => ({ wsOrigins: { value: null as string[] | null } }));
vi.mock('../../../src/app-config', async (importOriginal) => {
const actual = (await importOriginal()) as Record<string, unknown>;
return {
...actual,
readEnv: () => ({
...(actual.readEnv as () => { http: Record<string, unknown> })(),
http: {
...((actual.readEnv as () => { http: Record<string, unknown> })().http),
wsOrigins: wsOrigins.value,
},
}),
};
});
import { TrekWsAdapter } from '../../../src/nest/realtime/trek-ws.adapter';
import { getServer } from '../../../src/nest/realtime/ws-state';
import type { Server as HttpServer } from 'node:http';
type MessageListener = (buffer: Buffer) => void;
function fakeSocket() {
const sent: string[] = [];
const listeners: Record<string, MessageListener[]> = {};
return {
readyState: 1,
sent,
terminate: vi.fn(),
send: (raw: string) => { sent.push(raw); },
on: (event: string, fn: MessageListener) => {
(listeners[event] ??= []).push(fn);
},
emit: (event: string, payload: Buffer) => {
for (const fn of listeners[event] ?? []) fn(payload);
},
listenerCount: (event: string) => (listeners[event] ?? []).length,
};
}
const adapter = new TrekWsAdapter({} as HttpServer);
const frame = (o: unknown) => Buffer.from(JSON.stringify(o));
/** The adapter hands results to Nest's transform; here it is identity. */
const transform = (v: unknown) => ({ subscribe: (o: { next: (x: unknown) => void }) => o.next(v) }) as never;
let handled: unknown[];
let handlers: { message: string; callback: (data: unknown, socket: unknown) => unknown }[];
beforeEach(() => {
handled = [];
handlers = [
{ message: 'join', callback: (data) => { handled.push(data); return { type: 'joined' }; } },
{ message: 'leave', callback: (data) => { handled.push(data); return undefined; } },
];
});
describe('TrekWsAdapter wire protocol', () => {
it('WSAD-001: dispatches on `type`, and hands the WHOLE frame to the handler', () => {
// TREK's frames are flat: tripId sits beside type, not under a `data` key.
// The stock adapter reads `message.event` and passes `message.data`, which
// would leave join unhandled and the payload undefined.
const socket = fakeSocket();
adapter.bindMessageHandlers(socket as never, handlers as never, transform);
socket.emit('message', frame({ type: 'join', tripId: 7 }));
expect(handled).toEqual([{ type: 'join', tripId: 7 }]);
expect(JSON.parse(socket.sent[0])).toEqual({ type: 'joined' });
});
it('WSAD-001b: a socket that closed mid-dispatch is not written to', () => {
const socket = fakeSocket();
adapter.bindMessageHandlers(socket as never, handlers as never, transform);
(socket as { readyState: number }).readyState = 3; // CLOSED
socket.emit('message', frame({ type: 'join', tripId: 7 }));
expect(handled).toHaveLength(1); // the handler still ran
expect(socket.sent).toHaveLength(0); // the answer went nowhere
});
it('WSAD-002: a handler that answers undefined sends nothing', () => {
const socket = fakeSocket();
adapter.bindMessageHandlers(socket as never, handlers as never, transform);
socket.emit('message', frame({ type: 'leave', tripId: 7 }));
expect(socket.sent).toHaveLength(0);
});
it('WSAD-003: malformed JSON and unknown types are ignored, not answered', () => {
const socket = fakeSocket();
adapter.bindMessageHandlers(socket as never, handlers as never, transform);
socket.emit('message', Buffer.from('{not json'));
socket.emit('message', frame({ type: 'nope' }));
socket.emit('message', frame({ tripId: 7 }));
socket.emit('message', frame('a string'));
expect(handled).toHaveLength(0);
expect(socket.sent).toHaveLength(0);
});
});
describe('TrekWsAdapter flood guard', () => {
it('WSAD-010: counts EVERY frame, including the ones no handler wants', () => {
// The guard sits before parsing on purpose. Counting only the frames that
// reach a handler would let a client flood the process with garbage for
// free, which is what the limit exists to stop.
const socket = fakeSocket();
adapter.bindMessageHandlers(socket as never, handlers as never, transform);
for (let i = 0; i < 30; i++) socket.emit('message', Buffer.from('{not json'));
expect(socket.sent).toHaveLength(0);
socket.emit('message', frame({ type: 'join', tripId: 7 }));
expect(JSON.parse(socket.sent[0])).toEqual({ type: 'error', message: 'Rate limit exceeded' });
expect(handled).toHaveLength(0);
});
it('WSAD-010b: a socket that closed while flooding is not written to either', () => {
// The refusal is still a write. A client that hangs up mid-flood would
// otherwise take an ERR_HTTP_HEADERS-shaped error on the way out.
const socket = fakeSocket();
adapter.bindMessageHandlers(socket as never, handlers as never, transform);
for (let i = 0; i < 30; i++) socket.emit('message', Buffer.from('{not json'));
(socket as { readyState: number }).readyState = 3; // CLOSED
socket.emit('message', frame({ type: 'join', tripId: 7 }));
expect(socket.sent).toHaveLength(0);
});
it('WSAD-011: the window reopens, so a well-behaved client recovers', () => {
vi.useFakeTimers();
try {
const socket = fakeSocket();
adapter.bindMessageHandlers(socket as never, handlers as never, transform);
for (let i = 0; i < 31; i++) socket.emit('message', frame({ type: 'join', tripId: 7 }));
expect(socket.sent.some((s) => s.includes('Rate limit'))).toBe(true);
vi.advanceTimersByTime(10_001);
socket.sent.length = 0;
socket.emit('message', frame({ type: 'join', tripId: 7 }));
expect(JSON.parse(socket.sent[0])).toEqual({ type: 'joined' });
} finally {
vi.useRealTimers();
}
});
});
describe('TrekWsAdapter connection binding', () => {
it('WSAD-020: attaches the error listener BEFORE handing the socket on', () => {
// A protocol violation (a malformed frame, a reserved close code such as
// 1006) arrives as an `error` event; unhandled, Node rethrows it and the
// process dies. That is #1576, and it only reproduces against a hostile
// client, so nothing in CI would notice this listener going missing.
const socket = fakeSocket();
const server = {
on: (_event: string, cb: (s: unknown, r: unknown) => void) => cb(socket, { url: '/ws' }),
};
let listenersWhenCalled = -1;
adapter.bindClientConnect(server as never, () => {
listenersWhenCalled = socket.listenerCount('error');
});
expect(listenersWhenCalled).toBe(1);
});
it('WSAD-021: forwards the upgrade request, which carries the ws token', () => {
const socket = fakeSocket();
const request = { url: '/ws?token=abc' };
const server = {
on: (_event: string, cb: (s: unknown, r: unknown) => void) => cb(socket, request),
};
const seen: unknown[] = [];
adapter.bindClientConnect(server as never, (...args: unknown[]) => seen.push(...args));
expect(seen[1]).toBe(request);
});
});
describe('TrekWsAdapter server creation', () => {
/** A real http.Server is never listened on; ws only needs it to hang an upgrade handler. */
const httpServer = { on: vi.fn(), once: vi.fn(), removeListener: vi.fn(), emit: vi.fn() } as unknown as HttpServer;
it('WSAD-030: attaches to the http server it was given, on the gateway path', () => {
const ad = new TrekWsAdapter(httpServer);
const server = ad.create(0, { path: '/ws' }) as { options: Record<string, unknown> };
try {
expect(server.options.path).toBe('/ws');
// The registry has to see the live server, or every bridge broadcast
// silently goes nowhere.
expect(getServer()).toBe(server);
} finally {
void ad.close(server as never);
}
});
it('WSAD-031: with no allowlist configured, ws gets no verifyClient at all', () => {
// Rather than one that always says yes: an always-true hook is a hook that
// can be broken into a false later without anyone noticing.
wsOrigins.value = null;
const ad = new TrekWsAdapter(httpServer);
const server = ad.create(0, { path: '/ws' }) as { options: { verifyClient?: unknown } };
try {
expect(typeof server.options.verifyClient).not.toBe('function');
} finally {
void ad.close(server as never);
}
});
it('WSAD-031b: with an allowlist, a foreign origin is refused 403 before a socket exists', () => {
wsOrigins.value = ['https://trip.example'];
const ad = new TrekWsAdapter(httpServer);
const server = ad.create(0, { path: '/ws' }) as { options: { verifyClient?: unknown } };
try {
const verify = server.options.verifyClient as (
info: { origin: string },
cb: (ok: boolean, code?: number, msg?: string) => void,
) => void;
const seen: unknown[][] = [];
verify({ origin: 'https://evil.example' }, (...args) => seen.push(args));
verify({ origin: 'https://trip.example' }, (...args) => seen.push(args));
// A same-origin upgrade sends no Origin header at all; refusing that would
// lock out every non-browser client.
verify({ origin: '' }, (...args) => seen.push(args));
expect(seen).toEqual([[false, 403, 'Origin not allowed'], [true], [true]]);
} finally {
void ad.close(server as never);
}
});
it('WSAD-030b: falls back to /ws when the gateway declares no path', () => {
wsOrigins.value = null;
const ad = new TrekWsAdapter(httpServer);
const server = ad.create(0, {}) as { options: Record<string, unknown> };
try {
expect(server.options.path).toBe('/ws');
} finally {
void ad.close(server as never);
}
});
it('WSAD-031c: a server-level error is absorbed, not rethrown at the process', () => {
wsOrigins.value = null;
const ad = new TrekWsAdapter(httpServer);
const server = ad.create(0, { path: '/ws' }) as { emit: (e: string, err: Error) => boolean };
try {
expect(() => server.emit('error', new Error('upstream reset'))).not.toThrow();
} finally {
void ad.close(server as never);
}
});
it('WSAD-032: closing clears the registry, so a stale server is never broadcast to', async () => {
const ad = new TrekWsAdapter(httpServer);
const server = ad.create(0, { path: '/ws' });
await ad.close(server as never);
expect(getServer()).toBeNull();
});
});
server/tests/websocket/connection.test.ts
+4 -4
View File
@@ -48,13 +48,12 @@ vi.mock('../../src/config', () => ({
}));
import type { INestApplication } from '@nestjs/common';
import { buildApp } from '../../src/bootstrap';
import { buildApp, getHttpServer } from '../../src/bootstrap';
import { createTables } from '../../src/db/schema';
import { runMigrations } from '../../src/db/migrations';
import { resetTestDb, resetRateLimits } from '../helpers/test-db';
import { createUser, createTrip } from '../helpers/factories';
import { authCookie } from '../helpers/auth';
import { setupWebSocket } from '../../src/websocket';
import { createEphemeralToken } from '../../src/nest/auth/ephemeral-tokens';
import { createWsToken } from '../../src/nest/auth/auth.bridge';
@@ -69,8 +68,9 @@ beforeAll(async () => {
// Real WebSocket against the unified NestJS app (Express is gone). buildApp owns
// the same composition production uses; we attach the real ws server to it.
nestApp = await buildApp();
server = http.createServer(nestApp.getHttpAdapter().getInstance());
setupWebSocket(server);
// buildApp owns the server now, because the ws gateway is bound to it before
// app.init(). Creating a second one here would leave /ws unreachable.
server = getHttpServer();
await new Promise<void>(resolve => server.listen(0, resolve));
const addr = server.address() as { port: number };
server/tests/websocket/wire-envelope.test.ts
+4 -4
View File
@@ -50,12 +50,12 @@ vi.mock('../../src/config', () => ({
}));
import type { INestApplication } from '@nestjs/common';
import { buildApp } from '../../src/bootstrap';
import { buildApp, getHttpServer } from '../../src/bootstrap';
import { createTables } from '../../src/db/schema';
import { runMigrations } from '../../src/db/migrations';
import { resetTestDb } from '../helpers/test-db';
import { createUser, createTrip } from '../helpers/factories';
import { setupWebSocket, broadcast, broadcastToUser } from '../../src/websocket';
import { broadcast, broadcastToUser } from '../../src/websocket';
import { createEphemeralToken } from '../../src/nest/auth/ephemeral-tokens';
let server: http.Server;
@@ -66,8 +66,8 @@ beforeAll(async () => {
createTables(testDb);
runMigrations(testDb);
nestApp = await buildApp();
server = http.createServer(nestApp.getHttpAdapter().getInstance());
setupWebSocket(server);
// buildApp binds /ws to the server it creates, before app.init().
server = getHttpServer();
await new Promise<void>(resolve => server.listen(0, resolve));
const addr = server.address() as { port: number };
wsUrl = `ws://127.0.0.1:${addr.port}/ws`;