dandri/server
1
0
Fork 0
mirror of https://github.com/spacebarchat/server.git synced 2026-09-02 04:49:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Decouple gateway connection state from inner ws object

Browse Source
This commit is contained in:
Rory&
2026-08-20 12:32:28 +02:00
parent e1153f43f5
commit 5b3f180a9c
21 changed files with 143 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gateway/events/Close.ts
+26 -26
View File
@@ -22,41 +22,41 @@ import { WebSocket } from "@spacebar/gateway/util";
import { emitEvent, PresenceUpdateEvent, SessionsReplace, VoiceStateUpdateEvent, distributePresenceUpdate } from "@spacebar/util";
import { ProcessLifecycle } from "@spacebar/util/util/ProcessLifecycle";
export async function Close(this: WebSocket, code: number, reason: Buffer) {
export async function Close(socket: WebSocket, code: number, reason: Buffer) {
console.log("[WebSocket] closed", code, reason.toString());
if (this.heartbeatTimeout) clearTimeout(this.heartbeatTimeout);
if (this.readyTimeout) clearTimeout(this.readyTimeout);
this.deflate?.close();
this.inflate?.close();
this.removeAllListeners();
if (socket.heartbeatTimeout) clearTimeout(socket.heartbeatTimeout);
if (socket.readyTimeout) clearTimeout(socket.readyTimeout);
socket.deflate?.close();
socket.inflate?.close();
socket.rawSocket.removeAllListeners();
if (this.session) {
const authSessionId = this.session?.session_id;
if (socket.session) {
const authSessionId = socket.session?.session_id;
const closedAt = Date.now();
if (!(ProcessLifecycle.state === "stopping" || ProcessLifecycle.state === "stopped"))
setTimeout(async () => {
console.log("Handling presence update after disconnect");
try {
if (authSessionId && this.user_id) {
if (authSessionId && socket.user_id) {
const s = await Session.findOne({
where: { user_id: this.user_id, session_id: authSessionId },
where: { user_id: socket.user_id, session_id: authSessionId },
});
if (s && (s.last_seen?.getTime() ?? 0) <= closedAt) {
console.log("... updating session");
await Session.update({ user_id: this.user_id, session_id: authSessionId }, { status: "offline", activities: [], client_status: {} });
this.session = await Session.findOneOrFail({ where: { session_id: this.session_id } });
await Session.update({ user_id: socket.user_id, session_id: authSessionId }, { status: "offline", activities: [], client_status: {} });
socket.session = await Session.findOneOrFail({ where: { session_id: socket.session_id } });
console.log("... distributing PRESENCE_UPDATE");
await distributePresenceUpdate(this.user_id, {
await distributePresenceUpdate(socket.user_id, {
event: "PRESENCE_UPDATE",
data: {
user: (await User.findOneOrFail({ where: { id: this.user_id } })).toPublicUser(),
status: this.session!.getPublicStatus(),
client_status: this.session!.client_status,
activities: this.session!.activities,
user: (await User.findOneOrFail({ where: { id: socket.user_id } })).toPublicUser(),
status: socket.session!.getPublicStatus(),
client_status: socket.session!.client_status,
activities: socket.session!.activities,
},
origin: "GATEWAY_CLOSE",
transaction_id: `IDENT_${this.user_id}_${Random.getString("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789", 6)}`,
transaction_id: `IDENT_${socket.user_id}_${Random.getString("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789", 6)}`,
} satisfies PresenceUpdateEvent);
console.log("... done!");
} else console.log("... Discarding presence update as the session reactivated");
@@ -66,13 +66,13 @@ export async function Close(this: WebSocket, code: number, reason: Buffer) {
}
}, 10_000);
if (!this.user_id) console.error("No user id in websocket???", this);
if (!socket.user_id) console.error("No user id in websocket???", socket);
const voiceState = await VoiceState.findOne({
where: { user_id: this.user_id },
where: { user_id: socket.user_id },
});
// clear the voice state for this session if user was in voice channel
if (voiceState && voiceState.session_id === this.session_id && voiceState.channel_id) {
if (voiceState && voiceState.session_id === socket.session_id && voiceState.channel_id) {
const prevGuildId = voiceState.guild_id;
const prevChannelId = voiceState.channel_id;
@@ -104,13 +104,13 @@ export async function Close(this: WebSocket, code: number, reason: Buffer) {
}
}
if (this.user_id) {
if (socket.user_id) {
const sessions = await Session.find({
where: { user_id: this.user_id },
where: { user_id: socket.user_id },
});
await emitEvent({
event: "SESSIONS_REPLACE",
user_id: this.user_id,
user_id: socket.user_id,
data: sessions.map((x) => x.toPrivateGatewayDeviceInfo()),
} as SessionsReplace);
const session = sessions[0] || {
@@ -119,13 +119,13 @@ export async function Close(this: WebSocket, code: number, reason: Buffer) {
status: "offline",
};
const user = await User.getPublicUser(this.user_id).catch(() => undefined);
const user = await User.getPublicUser(socket.user_id).catch(() => undefined);
// Special case: dont emit a presence update for deleted users
if (user !== undefined)
await emitEvent({
event: "PRESENCE_UPDATE",
user_id: this.user_id,
user_id: socket.user_id,
data: {
user: user,
activities: session.activities,
src/gateway/events/Connection.ts
+18 -25
View File
@@ -48,10 +48,12 @@ const openConnectionCount = Monitoring.attachMetric(
}),
);
export async function Connection(this: WS.Server, socket: WebSocket, request: IncomingMessage) {
export async function Connection(this: WS.Server, rawSocket: WS, request: IncomingMessage) {
const socket = new WebSocket(rawSocket);
openConnections.push(socket);
openConnectionCount.set(openConnections.length);
socket.on("close", () => {
socket.rawSocket.on("close", () => {
const index = openConnections.indexOf(socket);
if (index !== -1) openConnections.splice(index, 1);
openConnectionCount.set(openConnections.length);
@@ -64,15 +66,15 @@ export async function Connection(this: WS.Server, socket: WebSocket, request: In
d: Math.round(Math.random() * 5000),
});
const closeListeners = socket.listeners("close");
const closeListeners = socket.rawSocket.listeners("close");
for (const listener of closeListeners) {
socket.off("close", listener);
socket.rawSocket.off("close", listener);
// noinspection JSVoidFunctionReturnValueUsed - awaiting results
const res = listener.call(socket, 1000, 0) as void | Promise<void>;
if (res) await res;
}
socket.close(1000);
socket.rawSocket.close(1000);
};
if (ProcessLifecycle.state == "stopping" || ProcessLifecycle.state == "stopped") return await onShutdown();
@@ -86,12 +88,12 @@ export async function Connection(this: WS.Server, socket: WebSocket, request: In
if (!ipAddress && Config.get().security.cdnSignatureIncludeIp) {
console.error("Gateway connection rejected: No IP address found.");
return socket.close(CLOSECODES.Decode_error, "Gateway connection rejected: IP address is required.");
return socket.rawSocket.close(CLOSECODES.Decode_error, "Gateway connection rejected: IP address is required.");
}
if (!socket.userAgent && Config.get().security.cdnSignatureIncludeUserAgent) {
console.error("Gateway connection rejected: No User-Agent header found.");
return socket.close(CLOSECODES.Decode_error, "Gateway connection rejected: User-Agent header is required.");
return socket.rawSocket.close(CLOSECODES.Decode_error, "Gateway connection rejected: User-Agent header is required.");
}
if (request.headers.cookie?.split("; ").find((x) => x.startsWith("__sb_sessid="))) {
@@ -105,12 +107,9 @@ export async function Connection(this: WS.Server, socket: WebSocket, request: In
socket.session_id = "TEMP_" + genSessionId(); //Set the session of the WebSocket object
try {
// @ts-ignore
socket.on("close", Close);
// @ts-ignore
socket.on("message", Message);
socket.on("error", (err) => console.error(`[Gateway/${socket.user_id ?? socket.ipAddress}]`, err));
socket.rawSocket.on("close", (code, reason) => Close(socket, code, reason));
socket.rawSocket.on("message", (data, isBinary) => Message(socket, data as Buffer));
socket.rawSocket.on("error", (err) => console.error(`[Gateway/${socket.user_id ?? socket.ipAddress}]`, err));
console.log(`[Gateway] New connection from ${ipAddress}, total ${this.clients.size}`);
@@ -125,7 +124,7 @@ export async function Connection(this: WS.Server, socket: WebSocket, request: In
"pong",
"unexpected-response",
].forEach((x) => {
socket.on(x, (y) => console.log(x, y));
socket.rawSocket.on(x, (y) => console.log(x, y));
});
const { searchParams } = new URL(`http://localhost${request.url}`);
@@ -133,13 +132,13 @@ export async function Connection(this: WS.Server, socket: WebSocket, request: In
socket.encoding = searchParams.get("encoding") || "json";
if (!["json", "etf"].includes(socket.encoding)) {
console.error(`[Gateway/${socket.ipAddress}] Unknown encoding: ${socket.encoding}`);
return socket.close(CLOSECODES.Decode_error);
return socket.rawSocket.close(CLOSECODES.Decode_error);
}
socket.version = Number(searchParams.get("version")) || 8;
if (socket.version != 8) {
console.error(`[Gateway/${socket.ipAddress}] Invalid API version: ${socket.version}`);
return socket.close(CLOSECODES.Invalid_API_version);
return socket.rawSocket.close(CLOSECODES.Invalid_API_version);
}
// @ts-ignore
@@ -153,16 +152,10 @@ export async function Connection(this: WS.Server, socket: WebSocket, request: In
socket.zstdDecoder = new Decoder();
} else {
console.error(`[Gateway/${socket.user_id}] Unknown compression: ${socket.compress}`);
return socket.close(CLOSECODES.Decode_error);
return socket.rawSocket.close(CLOSECODES.Decode_error);
}
}
socket.recentTransactions = [];
socket.events = {};
socket.member_events = {};
socket.permissions = {};
socket.sequence = 0;
setHeartbeat(socket);
await Send(socket, {
@@ -172,9 +165,9 @@ export async function Connection(this: WS.Server, socket: WebSocket, request: In
},
});
socket.readyTimeout = setTimeout(() => socket.close(CLOSECODES.Session_timed_out), 1000 * 30);
socket.readyTimeout = setTimeout(() => socket.rawSocket.close(CLOSECODES.Session_timed_out), 1000 * 30);
} catch (error) {
console.error(error);
return socket.close(CLOSECODES.Unknown_error);
return socket.rawSocket.close(CLOSECODES.Unknown_error);
}
}
src/gateway/events/Message.ts
+23 -23
View File
@@ -28,7 +28,7 @@ import { PayloadSchema } from "@spacebar/schemas";
const bigIntJson = BigIntJson({ storeAsString: true });
export async function Message(this: WebSocket, buffer: WS.Data) {
export async function Message(socket: WebSocket, buffer: WS.Data) {
// TODO: compression
let data: Payload;
@@ -37,60 +37,60 @@ export async function Message(this: WebSocket, buffer: WS.Data) {
typeof buffer === "string"
) {
data = bigIntJson.parse(buffer.toString());
} else if (this.encoding === "json" && Buffer.isBuffer(buffer)) {
if (this.compress === "zlib-stream") {
} else if (socket.encoding === "json" && Buffer.isBuffer(buffer)) {
if (socket.compress === "zlib-stream") {
try {
buffer = this.inflate!.process(buffer);
buffer = socket.inflate!.process(buffer);
} catch {
buffer = buffer.toString();
}
} else if (this.compress === "zstd-stream") {
} else if (socket.compress === "zstd-stream") {
try {
buffer = await this.zstdDecoder!.decode(buffer);
buffer = await socket.zstdDecoder!.decode(buffer);
} catch {
buffer = buffer.toString();
}
}
data = bigIntJson.parse(buffer as string);
} else if (this.encoding === "etf" && Buffer.isBuffer(buffer) && erlpack) {
} else if (socket.encoding === "etf" && Buffer.isBuffer(buffer) && erlpack) {
try {
// cast is ~safe: unpack returns the parsed data in the shape it was provided, @yukikaze-bot/erlpack got around this by returning `any` instead of an actual type union.
// cast is ~safe: unpack returns the parsed data in the shape it was provided, @yukikaze-bot/erlpack got around socket by returning `any` instead of an actual type union.
data = erlpack.unpack(buffer.buffer.slice(buffer.byteOffset, buffer.byteOffset + buffer.byteLength)) as unknown as Payload;
} catch {
console.error(`[Gateway/${this.user_id ?? this.ipAddress}] Failed to decode ETF payload`);
return this.close(CLOSECODES.Decode_error);
console.error(`[Gateway/${socket.user_id ?? socket.ipAddress}] Failed to decode ETF payload`);
return socket.rawSocket.close(CLOSECODES.Decode_error);
}
} else {
console.error(`[Gateway/${this.user_id ?? this.ipAddress}] Unknown payload format`);
return this.close(CLOSECODES.Decode_error);
console.error(`[Gateway/${socket.user_id ?? socket.ipAddress}] Unknown payload format`);
return socket.rawSocket.close(CLOSECODES.Decode_error);
}
if (process.env.WS_VERBOSE) console.log(`[Websocket] Incomming message: ${JSON.stringify(data)}`);
if (process.env.WS_VERBOSE) console.log(`[Websocket] Incoming message: ${JSON.stringify(data)}`);
if (process.env.WS_DUMP) {
const id = this.session_id || "unknown";
const id = socket.session_id || "unknown";
await fs.mkdir(path.join("dump", id), { recursive: true });
await fs.writeFile(path.join("dump", id, `${Date.now()}.in.json`), JSON.stringify(data, null, 2));
if (!this.session_id) console.log(`[Gateway/${this.user_id ?? this.ipAddress}] Unknown session id, dumping to unknown folder`);
if (!socket.session_id) console.log(`[Gateway/${socket.user_id ?? socket.ipAddress}] Unknown session id, dumping to unknown folder`);
}
check.call(this, PayloadSchema, data);
check.call(socket, PayloadSchema, data);
const OPCodeHandler = OPCodeHandlers[data.op];
if (!OPCodeHandler) {
console.error(`[Gateway/${this.user_id ?? this.ipAddress}] Unknown opcode`, data.op);
// TODO: if all opcodes are implemented comment this out:
// this.close(CLOSECODES.Unknown_opcode);
console.error(`[Gateway/${socket.user_id ?? socket.ipAddress}] Unknown opcode`, data.op);
// TODO: if all opcodes are implemented comment socket out:
// socket.close(CLOSECODES.Unknown_opcode);
return;
}
try {
return await OPCodeHandler.call(this, data);
return await OPCodeHandler.call(socket, data);
} catch (error) {
console.error(`[Gateway/${this.user_id ?? this.ipAddress}] Error: Op ${data.op}`, error);
// if (!this.CLOSED && this.CLOSING)
return this.close(CLOSECODES.Unknown_error);
console.error(`[Gateway/${socket.user_id ?? socket.ipAddress}] Error: Op ${data.op}`, error);
// if (!socket.CLOSED && socket.CLOSING)
return socket.rawSocket.close(CLOSECODES.Unknown_error);
}
}
src/gateway/listener/listener.ts
+4 -4
View File
@@ -139,7 +139,7 @@ export async function setupListener(this: WebSocket) {
} catch (e) {
console.error(`[RabbitMQ] [user-${this.user_id}] Failed to re-establish subscriptions:`, e);
// close the WebSocket - will force client to reconnect and redo subscription setup
this.close(4000, "Failed to re-establish event subscriptions");
this.rawSocket.close(4000, "Failed to re-establish event subscriptions");
}
};
@@ -156,7 +156,7 @@ export async function setupListener(this: WebSocket) {
RabbitMQ.on("reconnected", handleReconnect);
RabbitMQ.on("disconnected", handleDisconnect);
this.once("close", async () => {
this.rawSocket.once("close", async () => {
// Unsubscribe from RabbitMQ events
RabbitMQ.off("reconnected", handleReconnect);
RabbitMQ.off("disconnected", handleDisconnect);
@@ -208,7 +208,7 @@ async function consume(this: WebSocket, opts: EventOpts) {
s: this.sequence++,
d: opts.reconnect_delay ?? opts.data ?? 1000,
});
this.close(1000); // not a discord close code, standard WS "Normal Closure"
this.rawSocket.close(1000); // not a discord close code, standard WS "Normal Closure"
return;
case "SB_SESSION_REMOVE":
// TODO: what do we even send here?
@@ -216,7 +216,7 @@ async function consume(this: WebSocket, opts: EventOpts) {
op: OPCODES.Invalid_Session,
s: this.sequence++,
});
this.close(CLOSECODES.Invalid_session); // TODO: this is deprecated?
this.rawSocket.close(CLOSECODES.Invalid_session); // TODO: this is deprecated?
return;
default:
// no special treatment
src/gateway/opcodes/Identify.ts
+3 -3
View File
@@ -83,7 +83,7 @@ export async function onIdentify(this: WebSocket, data: Payload) {
if (this.user_id) {
// we've already identified
return this.close(CLOSECODES.Already_authenticated);
return this.rawSocket.close(CLOSECODES.Already_authenticated);
}
clearTimeout(this.readyTimeout);
@@ -111,7 +111,7 @@ export async function onIdentify(this: WebSocket, data: Payload) {
const user = tokenData.user;
if (!user) {
console.log(`[Gateway/${this.ipAddress}] Failed to identify user`);
return this.close(CLOSECODES.Authentication_failed);
return this.rawSocket.close(CLOSECODES.Authentication_failed);
}
this.user_id = user.id;
@@ -132,7 +132,7 @@ export async function onIdentify(this: WebSocket, data: Payload) {
if (this.shard_count == null || this.shard_id == null || this.shard_id > this.shard_count || this.shard_id < 0 || this.shard_count <= 0) {
// TODO: why do we even care about this right now?
console.log(`[Gateway/${this.user_id}] Invalid sharding from ${user.id}: ${identify.shard}`);
return this.close(CLOSECODES.Invalid_shard);
return this.rawSocket.close(CLOSECODES.Invalid_shard);
}
}
const validateIntentsAndShardingTime = taskSw.getElapsedAndReset();
src/gateway/opcodes/StreamCreate.ts
+1 -1
View File
@@ -49,7 +49,7 @@ export async function onStreamCreate(this: WebSocket, data: Payload) {
where: { id: body.channel_id },
});
if (!channel || (body.type === "guild" && channel.guild_id != body.guild_id)) return this.close(4000, "invalid channel");
if (!channel || (body.type === "guild" && channel.guild_id != body.guild_id)) return this.rawSocket.close(4000, "invalid channel");
// TODO: actually apply preferred_region from the event payload
const regions = Config.get().regions;
src/gateway/opcodes/StreamDelete.ts
+1 -1
View File
@@ -37,7 +37,7 @@ export async function onStreamDelete(this: WebSocket, data: Payload) {
try {
parsedKey = parseStreamKey(body.stream_key);
} catch (e) {
return this.close(4000, "Invalid stream key");
return this.rawSocket.close(4000, "Invalid stream key");
}
// noinspection JSUnusedLocalSymbols - TODO: what is type here?
src/gateway/opcodes/StreamWatch.ts
+4 -4
View File
@@ -40,7 +40,7 @@ export async function onStreamWatch(this: WebSocket, data: Payload) {
try {
parsedKey = parseStreamKey(body.stream_key);
} catch (e) {
return this.close(4000, "Invalid stream key");
return this.rawSocket.close(4000, "Invalid stream key");
}
const { type, channelId, guildId, userId } = parsedKey;
@@ -50,14 +50,14 @@ export async function onStreamWatch(this: WebSocket, data: Payload) {
relations: { channel: true },
});
if (!stream) return this.close(4000, "Invalid stream key");
if (!stream) return this.rawSocket.close(4000, "Invalid stream key");
if (type === "guild" && stream.channel.guild_id != guildId) return this.close(4000, "Invalid stream key");
if (type === "guild" && stream.channel.guild_id != guildId) return this.rawSocket.close(4000, "Invalid stream key");
const regions = Config.get().regions;
const guildRegion = regions.available.find((r) => r.endpoint === stream.endpoint);
if (!guildRegion) return this.close(4000, "Unknown region");
if (!guildRegion) return this.rawSocket.close(4000, "Unknown region");
const streamSession = StreamSession.create({
stream_id: stream.id,
src/gateway/opcodes/instanceOf.ts
+1 -1
View File
@@ -30,7 +30,7 @@ export function check(this: WebSocket, schema: unknown, data: unknown) {
} catch (error) {
console.error(error);
// invalid payload
this.close(CLOSECODES.Decode_error);
this.rawSocket.close(CLOSECODES.Decode_error);
throw error;
}
}
src/gateway/routes/_spacebar/gateway/admin/introspect.ts
+2 -1
View File
@@ -76,6 +76,7 @@ router.get(
open: openConnections.length,
sessions: openConnections.map((x) =>
// console.log(x);
// TODO: move to socket object
useFullWsObj
? {
...x,
@@ -86,7 +87,7 @@ router.get(
},
}
: {
wsReadystate: x.readyState,
wsReadystate: x.rawSocket.readyState,
version: x.version,
user_id: x.user_id,
session_id: x.session_id,
src/gateway/util/Heartbeat.ts
+1 -1
View File
@@ -23,5 +23,5 @@ import { WebSocket } from "./WebSocket";
export function setHeartbeat(socket: WebSocket) {
if (socket.heartbeatTimeout) clearTimeout(socket.heartbeatTimeout);
socket.heartbeatTimeout = setTimeout(() => socket.close(CLOSECODES.Session_timed_out), 1000 * 45);
socket.heartbeatTimeout = setTimeout(() => socket.rawSocket.close(CLOSECODES.Session_timed_out), 1000 * 45);
}
src/gateway/util/Send.ts
+3 -3
View File
@@ -69,13 +69,13 @@ export async function Send(socket: WebSocket, data: Payload) {
}
return new Promise((res, rej) => {
if (socket.readyState !== 1) {
if (socket.rawSocket.readyState !== 1) {
// return rej("socket not open");
socket.close();
socket.rawSocket.close();
return;
}
socket.send(buffer, (err) => {
socket.rawSocket.send(buffer, (err) => {
if (err) return rej(err);
return res(null);
});
src/gateway/util/WebSocket.ts
+11 -6
View File
@@ -24,8 +24,13 @@ import { Intents, ListenEventOpts, Permissions } from "@spacebar/util";
import { QoSPayload } from "../opcodes/Heartbeat";
import { Capabilities } from "./Capabilities";
export interface WebSocket extends WS {
recentTransactions: string[];
export class WebSocket {
rawSocket: WS;
constructor(socket: WS) {
this.rawSocket = socket;
}
recentTransactions: string[] = [];
version: number;
user_id: string;
session_id: string;
@@ -44,10 +49,10 @@ export interface WebSocket extends WS {
heartbeatTimeout: NodeJS.Timeout;
readyTimeout: NodeJS.Timeout;
intents: Intents;
sequence: number;
permissions: Record<string, Permissions>;
events: Record<string, undefined | (() => Promise<unknown>)>;
member_events: Record<string, () => Promise<unknown>>;
sequence: number = 0;
permissions: Record<string, Permissions> = {};
events: Record<string, undefined | (() => Promise<unknown>)> = {};
member_events: Record<string, () => Promise<unknown>> = {};
listen_options: ListenEventOpts;
capabilities?: Capabilities;
large_threshold: number;
src/webrtc/events/Close.ts
+5 -6
View File
@@ -16,25 +16,24 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import { WebSocket } from "@spacebar/gateway";
import { mediaServer, Send, VoiceOPCodes, WebRtcWebSocket } from "@spacebar/webrtc";
export async function onClose(this: WebRtcWebSocket, code: number, reason: string) {
export async function onClose(socket: WebRtcWebSocket, code: number, reason: Buffer) {
console.log("[WebRTC] closed", code, reason.toString());
if (this.user_id && this.webRtcClient) {
const { voiceRoomId } = this.webRtcClient;
if (socket.user_id && socket.webRtcClient) {
const { voiceRoomId } = socket.webRtcClient;
const connectedClients = mediaServer.getClientsForRtcServer<WebRtcWebSocket>(voiceRoomId);
for (const client of connectedClients) {
await Send(client.websocket, {
op: VoiceOPCodes.CLIENTS_CONNECT,
d: {
user_id: this.user_id,
user_id: socket.user_id,
},
});
}
}
this.removeAllListeners();
socket.rawSocket.removeAllListeners();
}
src/webrtc/events/Connection.ts
+8 -7
View File
@@ -28,10 +28,11 @@ import { onMessage } from "./Message";
// TODO: specify rate limit in config
// TODO: check msg max size
export async function Connection(this: WS.Server, socket: WebRtcWebSocket, request: IncomingMessage) {
export async function Connection(this: WS.Server, rawSocket: WS, request: IncomingMessage) {
const socket = new WebRtcWebSocket(rawSocket);
try {
socket.on("close", onClose.bind(socket));
socket.on("message", onMessage.bind(socket));
socket.rawSocket.on("close", (code, reason) => onClose(socket, code, reason));
socket.rawSocket.on("message", (data, isBinary) => onMessage(socket, data as Buffer));
console.log("[WebRTC] new connection", request.url);
if (process.env.WS_LOGEVENTS) {
@@ -45,7 +46,7 @@ export async function Connection(this: WS.Server, socket: WebRtcWebSocket, reque
"pong",
"unexpected-response",
].forEach((x) => {
socket.on(x, (y) => console.log("[WebRTC]", x, y));
socket.rawSocket.on(x, (y) => console.log("[WebRTC]", x, y));
});
}
@@ -53,11 +54,11 @@ export async function Connection(this: WS.Server, socket: WebRtcWebSocket, reque
socket.encoding = "json";
socket.version = Number(searchParams.get("v")) || 9;
if (socket.version < 3) return socket.close(CLOSECODES.Unknown_error, "invalid version");
if (socket.version < 3) return socket.rawSocket.close(CLOSECODES.Unknown_error, "invalid version");
setHeartbeat(socket);
socket.readyTimeout = setTimeout(() => socket.close(CLOSECODES.Session_timed_out), 1000 * 30);
socket.readyTimeout = setTimeout(() => socket.rawSocket.close(CLOSECODES.Session_timed_out), 1000 * 30);
await Send(socket, {
op: VoiceOPCodes.HELLO,
@@ -67,6 +68,6 @@ export async function Connection(this: WS.Server, socket: WebRtcWebSocket, reque
});
} catch (error) {
console.error("[WebRTC]", error);
return socket.close(CLOSECODES.Unknown_error);
return socket.rawSocket.close(CLOSECODES.Unknown_error);
}
}
src/webrtc/events/Message.ts
+3 -3
View File
@@ -20,10 +20,10 @@ import { CLOSECODES } from "@spacebar/gateway";
import OPCodeHandlers from "../opcodes";
import { VoiceOPCodes, VoicePayload, WebRtcWebSocket } from "../util";
export async function onMessage(this: WebRtcWebSocket, buffer: Buffer) {
export async function onMessage(socket: WebRtcWebSocket, buffer: Buffer) {
try {
const data: VoicePayload = JSON.parse(buffer.toString());
if (data.op !== VoiceOPCodes.IDENTIFY && !this.user_id) return this.close(CLOSECODES.Not_authenticated);
if (data.op !== VoiceOPCodes.IDENTIFY && !socket.user_id) return socket.rawSocket.close(CLOSECODES.Not_authenticated);
const OPCodeHandler = OPCodeHandlers[data.op];
if (!OPCodeHandler) {
@@ -37,7 +37,7 @@ export async function onMessage(this: WebRtcWebSocket, buffer: Buffer) {
console.log("[WebRTC] Opcode " + VoiceOPCodes[data.op]);
}
return await OPCodeHandler.call(this, data);
return await OPCodeHandler.call(socket, data);
} catch (error) {
console.error("[WebRTC] error", error);
// if (!this.CLOSED && this.CLOSING) return this.close(CloseCodes.Unknown_error);
src/webrtc/opcodes/Heartbeat.ts
+1 -1
View File
@@ -21,7 +21,7 @@ import { VoiceOPCodes, VoicePayload, WebRtcWebSocket, Send } from "../util";
export async function onHeartbeat(this: WebRtcWebSocket, data: VoicePayload) {
setHeartbeat(this);
if (isNaN(data.d)) return this.close(CLOSECODES.Decode_error);
if (isNaN(data.d)) return this.rawSocket.close(CLOSECODES.Decode_error);
await Send(this, { op: VoiceOPCodes.HEARTBEAT_ACK, d: data.d });
}
src/webrtc/opcodes/Identify.ts
+4 -4
View File
@@ -64,14 +64,14 @@ export async function onIdentify(this: WebRtcWebSocket, data: VoicePayload) {
streamSession.used = true;
await streamSession.save();
this.once("close", async () => {
this.rawSocket.once("close", async () => {
await streamSession.remove();
});
}
}
// if it doesnt match any then not valid token
if (!authenticated) return this.close(CLOSECODES.Authentication_failed);
if (!authenticated) return this.rawSocket.close(CLOSECODES.Authentication_failed);
this.user_id = user_id;
this.session_id = session_id;
@@ -82,10 +82,10 @@ export async function onIdentify(this: WebRtcWebSocket, data: VoicePayload) {
try {
this.webRtcClient = await mediaServer.join(voiceRoomId, this.user_id, this, type!);
} catch (e) {
return this.close(4013);
return this.rawSocket.close(4013);
}
this.on("close", () => {
this.rawSocket.on("close", () => {
// ice-lite media server relies on this to know when the peer went away
mediaServer.onClientClose(this.webRtcClient!);
});
src/webrtc/opcodes/SelectProtocol.ts
+2 -1
View File
@@ -15,6 +15,7 @@
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import { SelectProtocolSchema, validateSchema } from "@spacebar/schemas";
import { mediaServer, Send, VoiceOPCodes, VoicePayload, WebRtcWebSocket } from "@spacebar/webrtc";
@@ -24,7 +25,7 @@ export async function onSelectProtocol(this: WebRtcWebSocket, payload: VoicePayl
const data = validateSchema("SelectProtocolSchema", payload.d) as SelectProtocolSchema;
// UDP protocol not currently supported. Maybe in the future?
if (data.protocol !== "webrtc") return this.close(4000, "only webrtc protocol supported currently");
if (data.protocol !== "webrtc") return this.rawSocket.close(4000, "only webrtc protocol supported currently");
const response = await mediaServer.onOffer(this.webRtcClient, data.sdp!, data.codecs ?? []);
src/webrtc/util/Send.ts
+21 -3
View File
@@ -1,3 +1,21 @@
/*
Spacebar: A FOSS re-implementation and extension of the Discord.com backend.
Copyright (C) 2025 Spacebar and Spacebar Contributors
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published
by the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import { JSONReplacer } from "@spacebar/util";
import { VoicePayload } from "./Constants";
import { WebRtcWebSocket } from "./WebRtcWebSocket";
@@ -12,13 +30,13 @@ export function Send(socket: WebRtcWebSocket, data: VoicePayload) {
else return;
return new Promise((res, rej) => {
if (socket.readyState !== 1) {
if (socket.rawSocket.readyState !== 1) {
// return rej("socket not open");
socket.close();
socket.rawSocket.close();
return;
}
socket.send(buffer, (err) => {
socket.rawSocket.send(buffer, (err) => {
if (err) return rej(err);
return res(null);
});
src/webrtc/util/WebRtcWebSocket.ts
+1 -1
View File
@@ -1,7 +1,7 @@
import { WebSocket } from "@spacebar/gateway";
import type { WebRtcClient } from "@spacebarchat/spacebar-webrtc-types";
export interface WebRtcWebSocket extends WebSocket {
export class WebRtcWebSocket extends WebSocket {
type: "guild-voice" | "dm-voice" | "stream";
webRtcClient?: WebRtcClient<WebRtcWebSocket>;
}