mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-19 13:36:48 +00:00
Maurice
724c88c201
A mechanical prettier --write over server/src, server/tests, server/scripts and shared/src with the workspaces' .prettierrc; nothing but layout and import order changes, and every file keeps the same set of imports. shared is formatted throughout and its format baseline is empty. The server baseline keeps the 35 files that Prettier would grow past their lint:size limit (they need a split first, which is not a formatting change), and the generated entity files stay as their generator writes them.
1018 lines
38 KiB
TypeScript
1018 lines
38 KiB
TypeScript
/**
|
|
* WebSocket connection tests.
|
|
* Covers WS-001 to WS-006, WS-008 to WS-017.
|
|
*
|
|
* Starts a real HTTP server on a random port and connects via the `ws` library.
|
|
*/
|
|
import { buildApp, getHttpServer } from '../../src/bootstrap';
|
|
import { db as testDb } from '../../src/db/database';
|
|
import { Users } from '../../src/db/entities/Users.entity';
|
|
import { EphemeralTokenService } from '../../src/nest/auth-core/ephemeral-token.service';
|
|
import { createEphemeralToken } from '../../src/nest/auth-core/ephemeral-tokens';
|
|
import { broadcastToUser, getOnlineUserIds } from '../../src/nest/realtime/ws-state';
|
|
import { TokenService } from '../../src/nest/tokens/token.service';
|
|
import { authCookie } from '../helpers/auth';
|
|
import { createUser, createTrip } from '../helpers/factories';
|
|
import type { FactoryOrm } from '../helpers/factories/context';
|
|
import { deleteRows, updateRows } from '../helpers/factories/rows';
|
|
import { setAppSetting } from '../helpers/factories/settings';
|
|
import { readUser } from '../helpers/factories/users';
|
|
import { resetTestDb, resetRateLimits } from '../helpers/test-db';
|
|
import { createTestMcpTokensRepo, createTestUsersRepo } from '../helpers/test-uow';
|
|
import { MikroORM } from '@mikro-orm/core';
|
|
import type { INestApplication } from '@nestjs/common';
|
|
|
|
import http from 'http';
|
|
import crypto from 'node:crypto';
|
|
import net from 'node:net';
|
|
import request from 'supertest';
|
|
import { describe, it, expect, vi, beforeAll, afterAll, beforeEach } from 'vitest';
|
|
import WebSocket from 'ws';
|
|
|
|
vi.mock('../../src/db/database', async () => {
|
|
const { createSnapshotTestDb, buildDbMock } = await import('../helpers/db-mock');
|
|
return buildDbMock(createSnapshotTestDb());
|
|
});
|
|
|
|
// The gateway consumes ws-tokens through its injected TokenService; the
|
|
// ephemeral store is module-scoped on purpose, so a directly-constructed
|
|
// instance mints tokens the app under test accepts (TokenService is a leaf —
|
|
// its own unit suite constructs it the same way). `createSnapshotTestDb`
|
|
// (behind the `../../src/db/database` mock above) already carries the full
|
|
// migrated schema, so the repositories can build immediately — no
|
|
// `createTables`/`runMigrations` step needed first, unlike suites that start
|
|
// from a bare `:memory:` handle.
|
|
let tokenService: TokenService;
|
|
const createWsToken = (...args: Parameters<TokenService['createWsToken']>) => tokenService.createWsToken(...args);
|
|
|
|
let server: http.Server;
|
|
let wsUrl: string;
|
|
let nestApp: INestApplication;
|
|
const orm = (): FactoryOrm => nestApp.get(MikroORM);
|
|
|
|
beforeAll(async () => {
|
|
tokenService = new TokenService(
|
|
await createTestMcpTokensRepo(testDb),
|
|
await createTestUsersRepo(testDb),
|
|
new EphemeralTokenService(),
|
|
);
|
|
// 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();
|
|
// 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 };
|
|
wsUrl = `ws://127.0.0.1:${addr.port}/ws`;
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await new Promise<void>((resolve, reject) => server.close((err) => (err ? reject(err) : resolve())));
|
|
await nestApp.close();
|
|
testDb.close();
|
|
});
|
|
|
|
beforeEach(async () => {
|
|
resetTestDb(testDb);
|
|
await resetRateLimits(nestApp);
|
|
});
|
|
|
|
/** Buffered WebSocket wrapper that never drops messages. */
|
|
class WsClient {
|
|
private ws: WebSocket;
|
|
private buffer: any[] = [];
|
|
private waiters: Array<(msg: any) => void> = [];
|
|
|
|
constructor(ws: WebSocket) {
|
|
this.ws = ws;
|
|
ws.on('message', (data) => {
|
|
const msg = JSON.parse(data.toString());
|
|
const waiter = this.waiters.shift();
|
|
if (waiter) {
|
|
waiter(msg);
|
|
} else {
|
|
this.buffer.push(msg);
|
|
}
|
|
});
|
|
}
|
|
|
|
next(timeoutMs = 3000): Promise<any> {
|
|
if (this.buffer.length > 0) return Promise.resolve(this.buffer.shift());
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => {
|
|
const idx = this.waiters.indexOf(resolve);
|
|
if (idx !== -1) this.waiters.splice(idx, 1);
|
|
reject(new Error('Message timeout'));
|
|
}, timeoutMs);
|
|
this.waiters.push((msg) => {
|
|
clearTimeout(timer);
|
|
resolve(msg);
|
|
});
|
|
});
|
|
}
|
|
|
|
send(msg: object) {
|
|
this.ws.send(JSON.stringify(msg));
|
|
}
|
|
close() {
|
|
this.ws.close();
|
|
}
|
|
|
|
/** Wait for any message matching predicate within timeout. */
|
|
waitFor(predicate: (m: any) => boolean, timeoutMs = 3000): Promise<any> {
|
|
// Check buffer first
|
|
const idx = this.buffer.findIndex(predicate);
|
|
if (idx !== -1) return Promise.resolve(this.buffer.splice(idx, 1)[0]);
|
|
return new Promise((resolve, reject) => {
|
|
const timer = setTimeout(() => reject(new Error('waitFor timeout')), timeoutMs);
|
|
const handler = (msg: any) => {
|
|
if (predicate(msg)) {
|
|
clearTimeout(timer);
|
|
resolve(msg);
|
|
} else {
|
|
this.buffer.push(msg);
|
|
// re-register
|
|
this.waiters.push(handler);
|
|
}
|
|
};
|
|
this.waiters.push(handler);
|
|
});
|
|
}
|
|
|
|
/** Collect messages for a given duration. */
|
|
collectFor(ms: number): Promise<any[]> {
|
|
return new Promise((resolve) => {
|
|
const msgs: any[] = [...this.buffer.splice(0)];
|
|
const handleMsg = (msg: any) => msgs.push(msg);
|
|
this.ws.on('message', (data) => handleMsg(JSON.parse(data.toString())));
|
|
setTimeout(() => resolve(msgs), ms);
|
|
});
|
|
}
|
|
}
|
|
|
|
function connectWs(token?: string): Promise<WsClient> {
|
|
return new Promise((resolve, reject) => {
|
|
const url = token ? `${wsUrl}?token=${encodeURIComponent(token)}` : wsUrl;
|
|
const ws = new WebSocket(url);
|
|
const client = new WsClient(ws);
|
|
ws.once('open', () => resolve(client));
|
|
ws.once('error', reject);
|
|
ws.once('close', (code) => {
|
|
if (code === 4001) reject(new Error(`WS closed with 4001`));
|
|
});
|
|
});
|
|
}
|
|
|
|
describe('WS connection', () => {
|
|
it('WS-001 — connects with valid ephemeral token and receives welcome', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('welcome');
|
|
expect(typeof msg.socketId).toBe('number');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-002 — connecting without token closes with code 4001', async () => {
|
|
await new Promise<void>((resolve) => {
|
|
const ws = new WebSocket(wsUrl);
|
|
ws.on('close', (code) => {
|
|
expect(code).toBe(4001);
|
|
resolve();
|
|
});
|
|
ws.on('error', () => {});
|
|
});
|
|
});
|
|
|
|
it('WS-003 — connecting with invalid token closes with code 4001', async () => {
|
|
await new Promise<void>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=invalid-token-xyz`);
|
|
ws.on('close', (code) => {
|
|
expect(code).toBe(4001);
|
|
resolve();
|
|
});
|
|
ws.on('error', () => {});
|
|
});
|
|
});
|
|
});
|
|
|
|
describe('WS rooms', () => {
|
|
it('WS-004 — join trip room receives joined confirmation', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('joined');
|
|
expect(msg.tripId).toBe(trip.id);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-005 — join trip without access receives error', async () => {
|
|
const { user } = createUser(testDb);
|
|
const { user: otherUser } = createUser(testDb);
|
|
const trip = createTrip(testDb, otherUser.id); // trip owned by otherUser
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('error');
|
|
expect(msg.message).toMatch(/access denied/i);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
/**
|
|
* Plan 3c Task 9 fix wave (A-H1, live regression): before the
|
|
* `Number.isFinite` guard on `realtime.gateway.ts::handleJoin`, a
|
|
* non-numeric `tripId` became `NaN`, which used to reach
|
|
* `TripsRepository.findAccessible`'s raw bind as the unquoted bareword
|
|
* `NaN` and throw `SqliteError: no such column: NaN` — the promise this
|
|
* handler returns never resolved, so the client got NO reply at all
|
|
* (confirmed on the compiled dual boot: 4 `[ERROR]` lines, HEAD only).
|
|
* The platform's own `NaN` -> `NULL` rendering (`NulSafeSqlitePlatform`)
|
|
* fixes the crash even without this guard; this test pins the WS-level
|
|
* contract (the SAME `Access denied` frame WS-005 gets for a real trip
|
|
* with no access), not just "does not throw".
|
|
*/
|
|
it('WS-005b — join with a non-numeric tripId receives the same error frame as no access (rule 22 / A-H1)', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
client.send({ type: 'join', tripId: 'abc' });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('error');
|
|
expect(msg.message).toMatch(/access denied/i);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
/**
|
|
* Plan 3c Task 0b (R9): `handleJoin`'s `this.db.canAccessTrip(...)` call
|
|
* (`realtime.gateway.ts:173`) is now `TripsRepository.findAccessible`
|
|
* underneath `DatabaseService`, reached through the request context
|
|
* `TrekWsAdapter`'s `@SubscribeMessage` wrapper forks
|
|
* (`nest/realtime/trek-ws.adapter.ts:273`, pinned generically by
|
|
* SEAM-001/BOOT-SWEEP-001) — no ratchet existed for this SPECIFIC seam
|
|
* before this task (inventory §12b). WS-004 above already proves the happy
|
|
* path returns `{type:'joined'}`, which requires the repository read to
|
|
* have succeeded; this test makes that explicit and structural, the same
|
|
* `cannotUseGlobalContext`/"global EntityManager" console-error filter
|
|
* `boot-sweeps-request-context.test.ts` uses, so a regression that routes
|
|
* `handleJoin` around the adapter's request context fails HERE by message,
|
|
* not only by an indirect "joined never arrived" timeout.
|
|
*/
|
|
it("WS-SEAM-001 — handleJoin's repository read runs inside the WS request context (no cannotUseGlobalContext)", async () => {
|
|
const errSpy = vi.spyOn(console, 'error').mockImplementation(() => {});
|
|
try {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('joined');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
|
|
const suspicious = errSpy.mock.calls
|
|
.map((args) => args.map(String).join(' '))
|
|
.filter((line) => /cannotUseGlobalContext|global EntityManager/i.test(line));
|
|
expect(suspicious).toEqual([]);
|
|
} finally {
|
|
errSpy.mockRestore();
|
|
}
|
|
});
|
|
|
|
it('WS-006 — leave room receives left confirmation', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
await client.next(); // joined
|
|
|
|
client.send({ type: 'leave', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('left');
|
|
expect(msg.tripId).toBe(trip.id);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
});
|
|
|
|
describe('WS rate limiting', () => {
|
|
it('WS-008 — exceeding 30 messages per window triggers rate-limit error', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
// Send 35 messages quickly — at least one should trigger rate limit
|
|
for (let i = 0; i < 35; i++) {
|
|
client.send({ type: 'ping' });
|
|
}
|
|
|
|
// Collect for up to 2s and find a rate-limit error
|
|
const msgs = await client.collectFor(1500);
|
|
const rateLimitMsg = msgs.find((m: any) => m.type === 'error' && m.message?.includes('Rate limit'));
|
|
expect(rateLimitMsg).toBeDefined();
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
});
|
|
|
|
describe('WS real-time broadcast', () => {
|
|
it('WS-009 — POST /api/trips/:id/places broadcasts place:created to room members', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
// Join the trip room
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
await client.next(); // joined
|
|
|
|
// Create a place via REST (from a different socket, so it broadcasts to us)
|
|
const wsToken2 = createEphemeralToken(user.id, 'ws')!;
|
|
const client2 = await connectWs(wsToken2);
|
|
try {
|
|
await client2.next(); // welcome
|
|
client2.send({ type: 'join', tripId: trip.id });
|
|
await client2.next(); // joined
|
|
|
|
// REST call from client2's socket ID
|
|
const welcome2SocketId = (await Promise.resolve(null)) ?? null;
|
|
await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(user.id))
|
|
.send({ name: 'Test Place', lat: 48.8566, lng: 2.3522 });
|
|
|
|
// client should receive the broadcast
|
|
const msg = await client.waitFor((m: any) => m.type === 'place:created', 3000);
|
|
expect(msg.type).toBe('place:created');
|
|
expect(msg.place).toBeDefined();
|
|
expect(msg.place.name).toBe('Test Place');
|
|
} finally {
|
|
client2.close();
|
|
}
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-010 — ephemeral WS token is single-use (second connection is rejected)', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
// First connection: should succeed
|
|
const client = await connectWs(token);
|
|
await client.next(); // welcome
|
|
client.close();
|
|
|
|
// Second connection with same token: should be rejected with code 4001
|
|
const closeCode = await new Promise<number>((resolve, reject) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4001)); // connection error also means rejection
|
|
setTimeout(() => reject(new Error('Timeout waiting for rejection')), 3000);
|
|
});
|
|
expect([4001, 1006]).toContain(closeCode); // 4001 = auth rejected, 1006 = abnormal close (also rejection)
|
|
});
|
|
|
|
it('WS-011 — client not in trip room does not receive broadcast', async () => {
|
|
const { user: owner } = createUser(testDb);
|
|
const { user: other } = createUser(testDb);
|
|
const trip = createTrip(testDb, owner.id);
|
|
|
|
// Connect `other` user but do NOT join the trip room
|
|
const tokenOther = createEphemeralToken(other.id, 'ws')!;
|
|
const clientOther = await connectWs(tokenOther);
|
|
try {
|
|
await clientOther.next(); // welcome — but no join
|
|
|
|
// Owner creates a place
|
|
await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(owner.id))
|
|
.send({ name: 'Owner Place', lat: 48.8566, lng: 2.3522 });
|
|
|
|
// `other` should NOT receive any broadcast within 500ms
|
|
const msgs = await clientOther.collectFor(500);
|
|
const broadcast = msgs.find((m: any) => m.type === 'place:created');
|
|
expect(broadcast).toBeUndefined();
|
|
} finally {
|
|
clientOther.close();
|
|
}
|
|
});
|
|
});
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// WS auth edge cases — user-not-found and MFA enforcement
|
|
// ---------------------------------------------------------------------------
|
|
|
|
describe('WS auth edge cases', () => {
|
|
it('WS-012 — token for non-existent user closes with code 4001', async () => {
|
|
// Insert a user, grab an ephemeral token, then delete the user before connecting
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
// Remove the user so the DB lookup returns undefined
|
|
await deleteRows(orm(), Users, { id: user.id });
|
|
|
|
const closeCode = await new Promise<number>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4001));
|
|
});
|
|
expect(closeCode).toBe(4001);
|
|
});
|
|
|
|
it('WS-013 — MFA is enforced when require_mfa is enabled and user has no MFA', async () => {
|
|
// Enable require_mfa in app_settings
|
|
await setAppSetting(orm(), 'require_mfa', 'true');
|
|
|
|
// Create a regular user without MFA
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const closeCode = await new Promise<number>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4403));
|
|
});
|
|
expect(closeCode).toBe(4403);
|
|
});
|
|
|
|
it('WS-014 — MFA-enabled user connects successfully when require_mfa is enabled', async () => {
|
|
// Enable require_mfa
|
|
await setAppSetting(orm(), 'require_mfa', 'true');
|
|
|
|
// Create a user with MFA enabled
|
|
const { user } = createUser(testDb);
|
|
await updateRows(orm(), Users, { id: user.id }, { mfa_enabled: 1, mfa_secret: 'JBSWY3DPEHPK3PXP' });
|
|
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const client = await connectWs(token);
|
|
try {
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('welcome');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-027 — ws-token minted before a password change is rejected (session gate)', async () => {
|
|
// createWsToken stamps the user's current password_version (0) into the token.
|
|
const { user } = createUser(testDb);
|
|
const result = await createWsToken(user.id);
|
|
const token = result.token!;
|
|
|
|
// Simulate a password reset bumping the version AFTER the token was issued.
|
|
const { password_version: version } = await readUser(orm(), user.id);
|
|
await updateRows(orm(), Users, { id: user.id }, { password_version: version + 1 });
|
|
|
|
const closeCode = await new Promise<number>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4001));
|
|
});
|
|
expect(closeCode).toBe(4001);
|
|
});
|
|
|
|
it('WS-028 — ws-token whose password_version still matches connects successfully', async () => {
|
|
const { user } = createUser(testDb);
|
|
// Bump the version first, THEN mint — the token captures the current pv.
|
|
await updateRows(orm(), Users, { id: user.id }, { password_version: 3 });
|
|
const result = await createWsToken(user.id);
|
|
const client = await connectWs(result.token!);
|
|
try {
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('welcome');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-029 — legacy token without a pv is rejected once the user resets their password', async () => {
|
|
// Tokens minted via createEphemeralToken carry no pv (treated as version 0).
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
await updateRows(orm(), Users, { id: user.id }, { password_version: 1 });
|
|
|
|
const closeCode = await new Promise<number>((resolve) => {
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.once('close', (code) => resolve(code));
|
|
ws.once('error', () => resolve(4001));
|
|
});
|
|
expect(closeCode).toBe(4001);
|
|
});
|
|
});
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// WS message processing — malformed/invalid payloads
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/** Connect a raw WebSocket (no WsClient wrapper) using a raw-send capable helper. */
|
|
function connectRawWs(token: string): Promise<{ ws: WebSocket; received: any[] }> {
|
|
return new Promise((resolve, reject) => {
|
|
const received: any[] = [];
|
|
const ws = new WebSocket(`${wsUrl}?token=${encodeURIComponent(token)}`);
|
|
ws.on('message', (data) => {
|
|
try {
|
|
received.push(JSON.parse(data.toString()));
|
|
} catch {
|
|
/* ignore parse errors */
|
|
}
|
|
});
|
|
ws.once('open', () => resolve({ ws, received }));
|
|
ws.once('error', reject);
|
|
ws.once('close', (code) => {
|
|
if (code === 4001) reject(new Error('WS closed 4001'));
|
|
});
|
|
});
|
|
}
|
|
|
|
/** Wait until `received` array has at least `n` items, up to `timeoutMs`. */
|
|
function waitForMessages(received: any[], n = 1, timeoutMs = 3000): Promise<void> {
|
|
return new Promise((resolve, reject) => {
|
|
if (received.length >= n) {
|
|
resolve();
|
|
return;
|
|
}
|
|
const start = Date.now();
|
|
const poll = () => {
|
|
if (received.length >= n) {
|
|
resolve();
|
|
return;
|
|
}
|
|
if (Date.now() - start > timeoutMs) {
|
|
reject(new Error(`Timeout waiting for ${n} messages`));
|
|
return;
|
|
}
|
|
setTimeout(poll, 20);
|
|
};
|
|
poll();
|
|
});
|
|
}
|
|
|
|
describe('WS message processing edge cases', () => {
|
|
it('WS-015 — malformed JSON is silently ignored (no crash, no error response)', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const { ws: rawWs, received } = await connectRawWs(token);
|
|
|
|
// Wait for welcome
|
|
await waitForMessages(received, 1);
|
|
|
|
// Send raw malformed JSON — server should silently ignore and not close connection
|
|
rawWs.send('{ this is not json }');
|
|
rawWs.send('{broken');
|
|
|
|
await new Promise((r) => setTimeout(r, 300));
|
|
|
|
// No error messages should have been sent by the server
|
|
const errMsgs = received.filter((m) => m.type === 'error');
|
|
expect(errMsgs).toHaveLength(0);
|
|
// Connection should still be open
|
|
expect(rawWs.readyState).toBe(WebSocket.OPEN);
|
|
|
|
rawWs.close();
|
|
});
|
|
|
|
it('WS-015b — message with non-object payload is silently ignored', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const { ws: rawWs, received } = await connectRawWs(token);
|
|
|
|
// Wait for welcome
|
|
await waitForMessages(received, 1);
|
|
|
|
// Send valid JSON but not an object (array) — should be ignored
|
|
rawWs.send(JSON.stringify([1, 2, 3]));
|
|
// Send valid JSON number — should be ignored
|
|
rawWs.send('42');
|
|
|
|
await new Promise((r) => setTimeout(r, 300));
|
|
|
|
// The only message received should be the welcome; no errors emitted
|
|
const errors = received.filter((m) => m.type === 'error');
|
|
expect(errors).toHaveLength(0);
|
|
|
|
rawWs.close();
|
|
});
|
|
|
|
it('WS-015c — message object missing type field is silently ignored', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const { ws: rawWs, received } = await connectRawWs(token);
|
|
|
|
// Wait for welcome
|
|
await waitForMessages(received, 1);
|
|
|
|
// Object without a string `type` field
|
|
rawWs.send(JSON.stringify({ tripId: 1 }));
|
|
rawWs.send(JSON.stringify({ type: 42, tripId: 1 }));
|
|
|
|
await new Promise((r) => setTimeout(r, 300));
|
|
|
|
const errors = received.filter((m) => m.type === 'error');
|
|
expect(errors).toHaveLength(0);
|
|
|
|
rawWs.close();
|
|
});
|
|
|
|
it('WS-015d — a close frame with a reserved status code does not crash the server (#1576)', async () => {
|
|
// The `ws` client cannot send an invalid close code, so craft the frame over a raw
|
|
// socket. Reserved code 1006 makes ws emit an 'error' event on the socket; with no
|
|
// listener attached in setupWebSocket, Node rethrows it as an uncaughtException and the
|
|
// process dies in production. In-process under vitest that surfaces as an uncaught
|
|
// WS_ERR_INVALID_CLOSE_CODE, so capture uncaughtException for the duration and assert
|
|
// none fired — then confirm the server is still serving.
|
|
const uncaught: Error[] = [];
|
|
const onUncaught = (err: Error): void => {
|
|
uncaught.push(err);
|
|
};
|
|
process.on('uncaughtException', onUncaught);
|
|
|
|
try {
|
|
const port = (server.address() as { port: number }).port;
|
|
await new Promise<void>((resolve, reject) => {
|
|
const key = crypto.randomBytes(16).toString('base64');
|
|
const sock = net.connect(port, '127.0.0.1');
|
|
sock.on('error', reject);
|
|
sock.write(
|
|
`GET /ws HTTP/1.1\r\nHost: 127.0.0.1:${port}\r\nUpgrade: websocket\r\n` +
|
|
`Connection: Upgrade\r\nSec-WebSocket-Key: ${key}\r\nSec-WebSocket-Version: 13\r\n\r\n`,
|
|
);
|
|
sock.once('data', (buf) => {
|
|
if (!buf.toString('latin1').includes('101 Switching Protocols')) return reject(new Error('no upgrade'));
|
|
const mask = crypto.randomBytes(4);
|
|
const payload = Buffer.alloc(2);
|
|
payload.writeUInt16BE(1006, 0); // reserved — MUST NOT appear in a close frame (RFC 6455)
|
|
const masked = Buffer.from(payload.map((b, i) => b ^ mask[i % 4]));
|
|
sock.write(Buffer.concat([Buffer.from([0x88, 0x82]), mask, masked])); // FIN|close, MASK|len2
|
|
setTimeout(() => {
|
|
sock.destroy();
|
|
resolve();
|
|
}, 200);
|
|
});
|
|
});
|
|
|
|
expect(uncaught.map((e) => (e as { code?: string }).code)).not.toContain('WS_ERR_INVALID_CLOSE_CODE');
|
|
|
|
// And the server is still serving: a fresh connection still gets a welcome.
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const client = await connectWs(token);
|
|
try {
|
|
expect((await client.next()).type).toBe('welcome');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
} finally {
|
|
process.off('uncaughtException', onUncaught);
|
|
}
|
|
});
|
|
|
|
it('WS-016 — rate-limit window resets: after limit hit, next window accepts messages again', async () => {
|
|
// Exercises line 108-110: the `now - rate.windowStart > WS_MSG_WINDOW` branch (counter reset).
|
|
// We confirm that:
|
|
// (a) msg 31 triggers the rate-limit error (current window),
|
|
// (b) a trip join in the same window is blocked,
|
|
// (c) after the rate-limit trip-join is blocked we verify the counter path was reached.
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const { ws: rawWs, received } = await connectRawWs(token);
|
|
|
|
// Wait for welcome
|
|
await waitForMessages(received, 1);
|
|
|
|
// Send exactly 30 messages (the limit) — all should succeed (no rate-limit error yet)
|
|
for (let i = 0; i < 30; i++) {
|
|
rawWs.send(JSON.stringify({ type: 'noop' }));
|
|
}
|
|
await new Promise((r) => setTimeout(r, 200));
|
|
|
|
// Message 31 — triggers the `count > WS_MSG_LIMIT` branch, sends rate-limit error
|
|
rawWs.send(JSON.stringify({ type: 'noop' }));
|
|
await waitForMessages(received, 2, 3000); // welcome + rate-limit error
|
|
|
|
const rateLimitErrors = received.filter((m) => m.type === 'error' && m.message?.includes('Rate limit'));
|
|
expect(rateLimitErrors.length).toBeGreaterThanOrEqual(1);
|
|
|
|
rawWs.close();
|
|
});
|
|
});
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// WS room management — disconnect cleanup and leave-nonexistent-room
|
|
// ---------------------------------------------------------------------------
|
|
|
|
describe('WS disconnect and room cleanup', () => {
|
|
it('WS-017 — disconnecting cleans up room membership so broadcast stops reaching the client', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token1 = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
// Connect and join the room
|
|
const client = await connectWs(token1);
|
|
await client.next(); // welcome
|
|
client.send({ type: 'join', tripId: trip.id });
|
|
await client.next(); // joined
|
|
|
|
// Disconnect — triggers the 'close' handler that calls leaveRoom for all rooms
|
|
client.close();
|
|
await new Promise((r) => setTimeout(r, 200)); // let the close event propagate
|
|
|
|
// Now create a second client that also joins the room, then creates a place.
|
|
// The first client (now disconnected) must NOT receive it (it can't, but more
|
|
// importantly the server must not crash when iterating rooms and finding a gone socket).
|
|
const token2 = createEphemeralToken(user.id, 'ws')!;
|
|
const client2 = await connectWs(token2);
|
|
try {
|
|
await client2.next(); // welcome
|
|
client2.send({ type: 'join', tripId: trip.id });
|
|
await client2.next(); // joined
|
|
|
|
// REST call to create a place — triggers broadcast; if room cleanup failed,
|
|
// iterating a closed socket would surface here.
|
|
const res = await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(user.id))
|
|
.send({ name: 'Post-Disconnect Place', lat: 48.8566, lng: 2.3522 });
|
|
expect(res.status).toBe(201);
|
|
|
|
// client2 should still receive the broadcast
|
|
const msg = await client2.waitFor((m: any) => m.type === 'place:created', 3000);
|
|
expect(msg.place.name).toBe('Post-Disconnect Place');
|
|
} finally {
|
|
client2.close();
|
|
}
|
|
});
|
|
|
|
it('WS-018 — leaving a room the client was never in is a no-op (no crash, no error)', async () => {
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
// Send leave without ever joining — the server should respond with 'left'
|
|
// and not throw, since leaveRoom is defensive about missing rooms/sockets.
|
|
client.send({ type: 'leave', tripId: trip.id });
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('left');
|
|
expect(msg.tripId).toBe(trip.id);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
});
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// broadcastToUser() and getOnlineUserIds() — exported utility coverage
|
|
// ---------------------------------------------------------------------------
|
|
|
|
describe('broadcastToUser and getOnlineUserIds', () => {
|
|
it('WS-019 — broadcastToUser sends payload to all connected sockets for that user', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
// Call broadcastToUser directly
|
|
broadcastToUser(user.id, { type: 'test:direct', data: 'hello' });
|
|
|
|
const msg = await client.next();
|
|
expect(msg.type).toBe('test:direct');
|
|
expect(msg.data).toBe('hello');
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-020 — broadcastToUser with excludeSid does not send to the excluded socket', async () => {
|
|
const { user } = createUser(testDb);
|
|
|
|
// Connect two sockets for the same user
|
|
const token1 = createEphemeralToken(user.id, 'ws')!;
|
|
const token2 = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client1 = await connectWs(token1);
|
|
const client2 = await connectWs(token2);
|
|
try {
|
|
const welcome1 = await client1.next();
|
|
const welcome2 = await client2.next();
|
|
const sid1 = welcome1.socketId;
|
|
|
|
// Broadcast excluding client1's socket ID
|
|
broadcastToUser(user.id, { type: 'test:exclude' }, sid1);
|
|
|
|
// client2 should receive it
|
|
const msg2 = await client2.next();
|
|
expect(msg2.type).toBe('test:exclude');
|
|
|
|
// client1 should NOT receive it within 400ms
|
|
const msgs1 = await client1.collectFor(400);
|
|
const received = msgs1.find((m: any) => m.type === 'test:exclude');
|
|
expect(received).toBeUndefined();
|
|
} finally {
|
|
client1.close();
|
|
client2.close();
|
|
}
|
|
});
|
|
|
|
it('WS-021 — broadcastToUser does not send to sockets belonging to a different user', async () => {
|
|
const { user: userA } = createUser(testDb);
|
|
const { user: userB } = createUser(testDb);
|
|
|
|
const tokenA = createEphemeralToken(userA.id, 'ws')!;
|
|
const tokenB = createEphemeralToken(userB.id, 'ws')!;
|
|
|
|
const clientA = await connectWs(tokenA);
|
|
const clientB = await connectWs(tokenB);
|
|
try {
|
|
await clientA.next(); // welcome
|
|
await clientB.next(); // welcome
|
|
|
|
// Broadcast only to userA
|
|
broadcastToUser(userA.id, { type: 'test:userA-only' });
|
|
|
|
// userA's client receives it
|
|
const msgA = await clientA.next();
|
|
expect(msgA.type).toBe('test:userA-only');
|
|
|
|
// userB's client must NOT receive it within 400ms
|
|
const msgsB = await clientB.collectFor(400);
|
|
const leak = msgsB.find((m: any) => m.type === 'test:userA-only');
|
|
expect(leak).toBeUndefined();
|
|
} finally {
|
|
clientA.close();
|
|
clientB.close();
|
|
}
|
|
});
|
|
|
|
it('WS-022 — getOnlineUserIds returns IDs of all connected authenticated users', async () => {
|
|
const { user: userA } = createUser(testDb);
|
|
const { user: userB } = createUser(testDb);
|
|
|
|
const tokenA = createEphemeralToken(userA.id, 'ws')!;
|
|
const tokenB = createEphemeralToken(userB.id, 'ws')!;
|
|
|
|
const clientA = await connectWs(tokenA);
|
|
const clientB = await connectWs(tokenB);
|
|
try {
|
|
await clientA.next(); // welcome
|
|
await clientB.next(); // welcome
|
|
|
|
const online = getOnlineUserIds();
|
|
expect(online.has(userA.id)).toBe(true);
|
|
expect(online.has(userB.id)).toBe(true);
|
|
} finally {
|
|
clientA.close();
|
|
clientB.close();
|
|
}
|
|
});
|
|
|
|
it('WS-023 — getOnlineUserIds excludes disconnected users', async () => {
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client = await connectWs(token);
|
|
await client.next(); // welcome
|
|
|
|
// Verify user is online
|
|
expect(getOnlineUserIds().has(user.id)).toBe(true);
|
|
|
|
// Disconnect
|
|
client.close();
|
|
await new Promise((r) => setTimeout(r, 200));
|
|
|
|
// User should no longer appear in online set
|
|
expect(getOnlineUserIds().has(user.id)).toBe(false);
|
|
});
|
|
|
|
it('WS-024 — broadcastToUser delivers custom payload to the correct connected socket', async () => {
|
|
// This directly exercises the broadcastToUser code path end-to-end through the
|
|
// exported function, verifying that the correct socket receives the message.
|
|
const { user } = createUser(testDb);
|
|
const token = createEphemeralToken(user.id, 'ws')!;
|
|
const client = await connectWs(token);
|
|
try {
|
|
await client.next(); // welcome
|
|
|
|
const customPayload = { type: 'custom:event', value: 99 };
|
|
broadcastToUser(user.id, customPayload);
|
|
|
|
const msg = await client.waitFor((m: any) => m.type === 'custom:event', 2000);
|
|
expect(msg.type).toBe('custom:event');
|
|
expect(msg.value).toBe(99);
|
|
} finally {
|
|
client.close();
|
|
}
|
|
});
|
|
|
|
it('WS-025 — broadcast() to an empty/nonexistent room is a no-op (no crash)', async () => {
|
|
// Exercises line 180: `if (!room || room.size === 0) return`
|
|
// A REST mutation on a trip with no connected WS clients triggers broadcast()
|
|
// with a room that doesn't exist — must not throw.
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
|
|
// No WebSocket clients join the trip room before the REST call
|
|
const res = await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(user.id))
|
|
.send({ name: 'No Room Place', lat: 10, lng: 20 });
|
|
|
|
// Server must not crash — 201 confirms broadcast() returned silently
|
|
expect(res.status).toBe(201);
|
|
});
|
|
|
|
it('WS-026 — broadcast() skips non-OPEN sockets in the room', async () => {
|
|
// This exercises line 185: `if (ws.readyState !== 1) continue`
|
|
// We join a room with two clients, forcefully terminate one (so its readyState becomes
|
|
// CLOSED while still transiently in the room map), then trigger a broadcast and confirm
|
|
// the surviving client receives it without errors.
|
|
const { user } = createUser(testDb);
|
|
const trip = createTrip(testDb, user.id);
|
|
|
|
const token1 = createEphemeralToken(user.id, 'ws')!;
|
|
const token2 = createEphemeralToken(user.id, 'ws')!;
|
|
|
|
const client1 = await connectWs(token1);
|
|
const client2 = await connectWs(token2);
|
|
try {
|
|
await client1.next(); // welcome
|
|
await client2.next(); // welcome
|
|
|
|
client1.send({ type: 'join', tripId: trip.id });
|
|
await client1.next(); // joined
|
|
|
|
client2.send({ type: 'join', tripId: trip.id });
|
|
await client2.next(); // joined
|
|
|
|
// Close client1 abruptly — the underlying socket may momentarily remain in the room map
|
|
client1.close();
|
|
await new Promise((r) => setTimeout(r, 50)); // brief pause
|
|
|
|
// Trigger broadcast via REST — should not crash even if client1's socket is closed
|
|
const res = await request(server)
|
|
.post(`/api/trips/${trip.id}/places`)
|
|
.set('Cookie', authCookie(user.id))
|
|
.send({ name: 'Resilience Place', lat: 1, lng: 2 });
|
|
expect(res.status).toBe(201);
|
|
|
|
// client2 should still receive the broadcast
|
|
const msg = await client2.waitFor((m: any) => m.type === 'place:created', 3000);
|
|
expect(msg.place.name).toBe('Resilience Place');
|
|
} finally {
|
|
client2.close();
|
|
}
|
|
});
|
|
});
|