File size: 5,347 Bytes
87cb242 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | import WebSocket from 'ws';
import { IncomingMessage } from 'http';
import { v4 as uuidv4 } from 'uuid';
import { verifyToken, extractToken } from '../utils/jwt';
import { safeSend, parseClientMessage, ClientMessage } from '../utils/protocol';
import { setOnline, setOffline, refreshPresence } from '../redis/presenceClient';
import { flushOfflineQueue } from '../redis/messageClient';
import { registry } from '../services/connectionRegistry';
import { buildIceServers } from '../config';
import { logger } from '../utils/logger';
import { handleSignaling } from './signalingHandler';
import { handleMessage } from './messageHandler';
import { handleFileRequest } from './fileHandler';
import { handleFriend } from './friendHandler';
export async function onConnection(
ws: WebSocket,
req: IncomingMessage
): Promise<void> {
// ── 1. Authenticate ──────────────────────────────────────────────────────
let uid: string;
try {
const url = new URL(req.url ?? '/', `http://${req.headers.host}`);
const token = extractToken(
req.headers.authorization,
url.searchParams.get('token') ?? undefined
);
const payload = verifyToken(token);
uid = payload.uid;
} catch (err) {
const msg = err instanceof Error ? err.message : 'Auth failed';
ws.send(JSON.stringify({ type: 'error', payload: { message: msg }, ts: Date.now() }));
ws.terminate();
return;
}
const socketId = uuidv4();
// ── 2. Register in process registry + Redis Bucket 1 ────────────────────
registry.add(socketId, uid, ws);
// Redis errors must NOT crash the connection handler
try {
await setOnline(uid, socketId);
} catch (err) {
logger.error('setOnline failed', { uid, error: String(err) });
}
logger.info('Client connected', { uid, socketId, total: registry.size() });
// ── 3. Send welcome + ICE server config ─────────────────────────────────
safeSend(ws, 'connected', { uid, socketId });
safeSend(ws, 'ice-servers', { iceServers: buildIceServers() });
// ── 4. Flush offline message queue (Bucket 2) ────────────────────────────
try {
const pending = await flushOfflineQueue(uid);
if (pending.length > 0) {
safeSend(ws, 'offline-flush', { messages: pending, count: pending.length });
logger.info('Flushed offline queue', { uid, count: pending.length });
}
} catch (err) {
logger.error('Offline flush failed', { uid, error: String(err) });
}
// ── 5. Message router ────────────────────────────────────────────────────
ws.on('message', (raw) => {
const msg = parseClientMessage(raw);
if (!msg) return; // Malformed — silently drop
registry.heartbeat(socketId); // Any message counts as a heartbeat
routeMessage(ws, uid, socketId, msg).catch((err: unknown) => {
logger.error('Message handler error', { uid, type: msg.type, error: String(err) });
});
});
// ── 6. Cleanup on disconnect ─────────────────────────────────────────────
ws.on('close', () => {
registry.remove(socketId);
setOffline(uid).catch((err: unknown) =>
logger.error('setOffline failed', { uid, error: String(err) })
);
logger.info('Client disconnected', { uid, socketId, total: registry.size() });
});
ws.on('error', (err) => {
logger.error('WebSocket error', { uid, socketId, error: err.message });
ws.terminate();
});
}
// ── Message router ────────────────────────────────────────────────────────────
async function routeMessage(
ws: WebSocket,
uid: string,
socketId: string,
msg: ClientMessage
): Promise<void> {
switch (msg.type) {
case 'heartbeat':
await refreshPresence(uid).catch(() => {}); // Best effort
break;
case 'offer':
case 'answer':
case 'ice-candidate':
await handleSignaling(ws, uid, msg);
break;
case 'message':
await handleMessage(ws, uid, msg);
break;
case 'get-upload-url':
await handleFileRequest(ws, uid, msg);
break;
case 'relay':
await handleRelay(ws, uid, msg);
break;
case 'friend':
await handleFriend(ws, uid, msg);
break;
default:
safeSend(ws, 'error', { message: `Unknown message type: ${msg.type}` });
}
}
// ── WebSocket relay fallback (when WebRTC ICE fails) ─────────────────────────
async function handleRelay(
_senderWs: WebSocket,
fromUid: string,
msg: ClientMessage
): Promise<void> {
const toUid = msg.payload.toUid as string | undefined;
if (!toUid || typeof toUid !== 'string') return;
const target = registry.getByUid(toUid);
if (!target) return;
safeSend(target.ws, 'relay', {
fromUid,
data: msg.payload.data,
}, msg.requestId);
}
|