mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-19 21:46:49 +00:00
Maurice
d981719941
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.
110 lines
3.5 KiB
TypeScript
110 lines
3.5 KiB
TypeScript
/**
|
|
* WebSocket test client helper.
|
|
*
|
|
* Usage:
|
|
* import http from 'http';
|
|
* import { getHttpServer } from '../../src/bootstrap';
|
|
* import { WsTestClient, getWsToken } from '../helpers/ws-client';
|
|
*
|
|
* let server: http.Server;
|
|
* let client: WsTestClient;
|
|
*
|
|
* beforeAll(async () => {
|
|
* const app = createApp();
|
|
* server = http.createServer(app);
|
|
* const server = getHttpServer();
|
|
* await new Promise<void>(res => server.listen(0, res));
|
|
* });
|
|
*
|
|
* afterAll(() => server.close());
|
|
*
|
|
* it('connects', async () => {
|
|
* const addr = server.address() as AddressInfo;
|
|
* const token = await getWsToken(addr.port, userId);
|
|
* client = new WsTestClient(`ws://localhost:${addr.port}/ws?token=${token}`);
|
|
* const msg = await client.waitForMessage('welcome');
|
|
* expect(msg.type).toBe('welcome');
|
|
* });
|
|
*/
|
|
|
|
import WebSocket from 'ws';
|
|
|
|
export interface WsMessage {
|
|
type: string;
|
|
[key: string]: unknown;
|
|
}
|
|
|
|
export class WsTestClient {
|
|
private ws: WebSocket;
|
|
private messageQueue: WsMessage[] = [];
|
|
private waiters: Array<{ type: string; resolve: (msg: WsMessage) => void; reject: (err: Error) => void }> = [];
|
|
|
|
constructor(url: string) {
|
|
this.ws = new WebSocket(url);
|
|
this.ws.on('message', (data: WebSocket.RawData) => {
|
|
try {
|
|
const msg = JSON.parse(data.toString()) as WsMessage;
|
|
const waiterIdx = this.waiters.findIndex(w => w.type === msg.type || w.type === '*');
|
|
if (waiterIdx >= 0) {
|
|
const waiter = this.waiters.splice(waiterIdx, 1)[0];
|
|
waiter.resolve(msg);
|
|
} else {
|
|
this.messageQueue.push(msg);
|
|
}
|
|
} catch { /* ignore malformed messages */ }
|
|
});
|
|
}
|
|
|
|
/** Wait for a message of the given type (or '*' for any). */
|
|
waitForMessage(type: string, timeoutMs = 5000): Promise<WsMessage> {
|
|
// Check if already in queue
|
|
const idx = this.messageQueue.findIndex(m => type === '*' || m.type === type);
|
|
if (idx >= 0) {
|
|
return Promise.resolve(this.messageQueue.splice(idx, 1)[0]);
|
|
}
|
|
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => {
|
|
const waiterIdx = this.waiters.findIndex(w => w.resolve === resolve);
|
|
if (waiterIdx >= 0) this.waiters.splice(waiterIdx, 1);
|
|
reject(new Error(`Timed out waiting for WS message type="${type}" after ${timeoutMs}ms`));
|
|
}, timeoutMs);
|
|
|
|
this.waiters.push({
|
|
type,
|
|
resolve: (msg) => { clearTimeout(timer); resolve(msg); },
|
|
reject,
|
|
});
|
|
});
|
|
}
|
|
|
|
/** Send a JSON message. */
|
|
send(msg: Record<string, unknown>): void {
|
|
this.ws.send(JSON.stringify(msg));
|
|
}
|
|
|
|
/** Close the connection. */
|
|
close(): void {
|
|
this.ws.close();
|
|
}
|
|
|
|
/** Wait for the connection to be open. */
|
|
waitForOpen(timeoutMs = 3000): Promise<void> {
|
|
if (this.ws.readyState === WebSocket.OPEN) return Promise.resolve();
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => reject(new Error('WS open timed out')), timeoutMs);
|
|
this.ws.once('open', () => { clearTimeout(timer); resolve(); });
|
|
this.ws.once('error', (err) => { clearTimeout(timer); reject(err); });
|
|
});
|
|
}
|
|
|
|
/** Wait for the connection to close. */
|
|
waitForClose(timeoutMs = 3000): Promise<number> {
|
|
if (this.ws.readyState === WebSocket.CLOSED) return Promise.resolve(1000);
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => reject(new Error('WS close timed out')), timeoutMs);
|
|
this.ws.once('close', (code) => { clearTimeout(timer); resolve(code); });
|
|
});
|
|
}
|
|
}
|