stale server sockets, passive non-initiators, and race conditions during peer connection setup. Fix users unable to see or hear each other in voice channels due to stale server sockets, passive non-initiators, and race conditions during peer connection setup. Server: - Close stale WebSocket connections sharing the same oderId in handleIdentify instead of letting them linger up to 45s - Make user_joined/user_left broadcasts identity-aware so duplicate sockets don't produce phantom join/leave events - Include serverIds in user_left payload for multi-room presence - Simplify findUserByOderId now that stale sockets are cleaned up Client - signaling: - Add fallback offer system with 1s timer for missed user_joined races - Add non-initiator takeover after 5s when the initiator fails to send an offer (NON_INITIATOR_GIVE_UP_MS) - Scope peerServerMap per signaling URL to prevent cross-server collisions - Add socket identity guards on all signaling event handlers - Replace canReusePeerConnection with hasActivePeerConnection and isPeerConnectionNegotiating with extended grace periods Client - peer connections: - Extract replaceUnusablePeer helper to deduplicate stale peer replacement in offer and ICE handlers - Add stale connectionstatechange guard to ignore events from replaced RTCPeerConnection instances - Use deterministic initiator election in peer recovery reconnects - Track createdAt on PeerData for staleness detection Client - presence: - Add multi-room presence tracking via presenceServerIds on User - Replace clearUsers + individual userJoined with syncServerPresence for atomic server roster updates - Make userLeft handle partial server removal instead of full eviction Documentation: - Add server-side connection hygiene, non-initiator takeover, and stale peer replacement sections to the realtime README
241 lines
6.9 KiB
TypeScript
241 lines
6.9 KiB
TypeScript
import { connectedUsers } from './state';
|
|
import { ConnectedUser } from './types';
|
|
import {
|
|
broadcastToServer,
|
|
findUserByOderId,
|
|
getServerIdsForOderId,
|
|
getUniqueUsersInServer,
|
|
isOderIdConnectedToServer
|
|
} from './broadcast';
|
|
import { authorizeWebSocketJoin } from '../services/server-access.service';
|
|
|
|
interface WsMessage {
|
|
[key: string]: unknown;
|
|
type: string;
|
|
}
|
|
|
|
function normalizeDisplayName(value: unknown, fallback = 'User'): string {
|
|
const normalized = typeof value === 'string' ? value.trim() : '';
|
|
|
|
return normalized || fallback;
|
|
}
|
|
|
|
function readMessageId(value: unknown): string | undefined {
|
|
if (typeof value !== 'string') {
|
|
return undefined;
|
|
}
|
|
|
|
const normalized = value.trim();
|
|
|
|
if (!normalized || normalized === 'undefined' || normalized === 'null') {
|
|
return undefined;
|
|
}
|
|
|
|
return normalized;
|
|
}
|
|
|
|
/** Sends the current user list for a given server to a single connected user. */
|
|
function sendServerUsers(user: ConnectedUser, serverId: string): void {
|
|
const users = getUniqueUsersInServer(serverId, user.oderId)
|
|
.map(cu => ({ oderId: cu.oderId, displayName: normalizeDisplayName(cu.displayName) }));
|
|
|
|
user.ws.send(JSON.stringify({ type: 'server_users', serverId, users }));
|
|
}
|
|
|
|
function handleIdentify(user: ConnectedUser, message: WsMessage, connectionId: string): void {
|
|
const newOderId = readMessageId(message['oderId']) ?? connectionId;
|
|
|
|
// Close stale connections from the same identity so offer routing
|
|
// always targets the freshest socket (e.g. after page refresh).
|
|
connectedUsers.forEach((existing, existingId) => {
|
|
if (existingId !== connectionId && existing.oderId === newOderId) {
|
|
console.log(`Closing stale connection for ${newOderId} (old=${existingId}, new=${connectionId})`);
|
|
|
|
try {
|
|
existing.ws.close();
|
|
} catch { /* already closing */ }
|
|
|
|
connectedUsers.delete(existingId);
|
|
}
|
|
});
|
|
|
|
user.oderId = newOderId;
|
|
user.displayName = normalizeDisplayName(message['displayName'], normalizeDisplayName(user.displayName));
|
|
connectedUsers.set(connectionId, user);
|
|
console.log(`User identified: ${user.displayName} (${user.oderId})`);
|
|
}
|
|
|
|
async function handleJoinServer(user: ConnectedUser, message: WsMessage, connectionId: string): Promise<void> {
|
|
const sid = readMessageId(message['serverId']);
|
|
|
|
if (!sid)
|
|
return;
|
|
|
|
const authorization = await authorizeWebSocketJoin(sid, user.oderId);
|
|
|
|
if (!authorization.allowed) {
|
|
user.ws.send(JSON.stringify({
|
|
type: 'access_denied',
|
|
serverId: sid,
|
|
reason: authorization.reason
|
|
}));
|
|
|
|
return;
|
|
}
|
|
|
|
const isNewConnectionMembership = !user.serverIds.has(sid);
|
|
const isNewIdentityMembership = isNewConnectionMembership && !isOderIdConnectedToServer(user.oderId, sid, connectionId);
|
|
|
|
user.serverIds.add(sid);
|
|
user.viewedServerId = sid;
|
|
connectedUsers.set(connectionId, user);
|
|
console.log(
|
|
`User ${normalizeDisplayName(user.displayName)} (${user.oderId}) joined server ${sid} `
|
|
+ `(newConnection=${isNewConnectionMembership}, newIdentity=${isNewIdentityMembership})`
|
|
);
|
|
|
|
sendServerUsers(user, sid);
|
|
|
|
if (isNewIdentityMembership) {
|
|
broadcastToServer(sid, {
|
|
type: 'user_joined',
|
|
oderId: user.oderId,
|
|
displayName: normalizeDisplayName(user.displayName),
|
|
serverId: sid
|
|
}, user.oderId);
|
|
}
|
|
}
|
|
|
|
function handleViewServer(user: ConnectedUser, message: WsMessage, connectionId: string): void {
|
|
const viewSid = readMessageId(message['serverId']);
|
|
|
|
if (!viewSid)
|
|
return;
|
|
|
|
user.viewedServerId = viewSid;
|
|
connectedUsers.set(connectionId, user);
|
|
console.log(`User ${normalizeDisplayName(user.displayName)} (${user.oderId}) viewing server ${viewSid}`);
|
|
|
|
sendServerUsers(user, viewSid);
|
|
}
|
|
|
|
function handleLeaveServer(user: ConnectedUser, message: WsMessage, connectionId: string): void {
|
|
const leaveSid = readMessageId(message['serverId']) ?? user.viewedServerId;
|
|
|
|
if (!leaveSid)
|
|
return;
|
|
|
|
user.serverIds.delete(leaveSid);
|
|
|
|
if (user.viewedServerId === leaveSid)
|
|
user.viewedServerId = undefined;
|
|
|
|
connectedUsers.set(connectionId, user);
|
|
|
|
const remainingServerIds = getServerIdsForOderId(user.oderId, connectionId);
|
|
|
|
if (remainingServerIds.includes(leaveSid)) {
|
|
return;
|
|
}
|
|
|
|
broadcastToServer(leaveSid, {
|
|
type: 'user_left',
|
|
oderId: user.oderId,
|
|
displayName: normalizeDisplayName(user.displayName),
|
|
serverId: leaveSid,
|
|
serverIds: remainingServerIds
|
|
}, user.oderId);
|
|
}
|
|
|
|
function forwardRtcMessage(user: ConnectedUser, message: WsMessage): void {
|
|
const targetUserId = readMessageId(message['targetUserId']) ?? '';
|
|
|
|
console.log(`Forwarding ${message.type} from ${user.oderId} to ${targetUserId}`);
|
|
|
|
const targetUser = findUserByOderId(targetUserId);
|
|
|
|
if (targetUser) {
|
|
targetUser.ws.send(JSON.stringify({ ...message, fromUserId: user.oderId }));
|
|
console.log(`Successfully forwarded ${message.type} to ${targetUserId}`);
|
|
} else {
|
|
console.log(
|
|
`Target user ${targetUserId} not found. Connected users:`,
|
|
Array.from(connectedUsers.values()).map(cu => ({ oderId: cu.oderId, displayName: cu.displayName }))
|
|
);
|
|
}
|
|
}
|
|
|
|
function handleChatMessage(user: ConnectedUser, message: WsMessage): void {
|
|
const chatSid = (message['serverId'] as string | undefined) ?? user.viewedServerId;
|
|
|
|
if (chatSid && user.serverIds.has(chatSid)) {
|
|
broadcastToServer(chatSid, {
|
|
type: 'chat_message',
|
|
serverId: chatSid,
|
|
message: message['message'],
|
|
senderId: user.oderId,
|
|
senderName: user.displayName,
|
|
timestamp: Date.now()
|
|
});
|
|
}
|
|
}
|
|
|
|
function handleTyping(user: ConnectedUser, message: WsMessage): void {
|
|
const typingSid = (message['serverId'] as string | undefined) ?? user.viewedServerId;
|
|
const channelId = typeof message['channelId'] === 'string' && message['channelId'].trim()
|
|
? message['channelId'].trim()
|
|
: 'general';
|
|
|
|
if (typingSid && user.serverIds.has(typingSid)) {
|
|
broadcastToServer(typingSid, {
|
|
type: 'user_typing',
|
|
serverId: typingSid,
|
|
channelId,
|
|
oderId: user.oderId,
|
|
displayName: user.displayName
|
|
}, user.oderId);
|
|
}
|
|
}
|
|
|
|
export async function handleWebSocketMessage(connectionId: string, message: WsMessage): Promise<void> {
|
|
const user = connectedUsers.get(connectionId);
|
|
|
|
if (!user)
|
|
return;
|
|
|
|
switch (message.type) {
|
|
case 'identify':
|
|
handleIdentify(user, message, connectionId);
|
|
break;
|
|
|
|
case 'join_server':
|
|
await handleJoinServer(user, message, connectionId);
|
|
break;
|
|
|
|
case 'view_server':
|
|
handleViewServer(user, message, connectionId);
|
|
break;
|
|
|
|
case 'leave_server':
|
|
handleLeaveServer(user, message, connectionId);
|
|
break;
|
|
|
|
case 'offer':
|
|
case 'answer':
|
|
case 'ice_candidate':
|
|
forwardRtcMessage(user, message);
|
|
break;
|
|
|
|
case 'chat_message':
|
|
handleChatMessage(user, message);
|
|
break;
|
|
|
|
case 'typing':
|
|
handleTyping(user, message);
|
|
break;
|
|
|
|
default:
|
|
console.log('Unknown message type:', message.type);
|
|
}
|
|
}
|