diff --git a/src/gateway/events/Close.ts b/src/gateway/events/Close.ts
index 0cdd5741..3f3e9475 100644
--- a/src/gateway/events/Close.ts
+++ b/src/gateway/events/Close.ts
@@ -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,
diff --git a/src/gateway/events/Connection.ts b/src/gateway/events/Connection.ts
index 32f4e14d..bd17628f 100644
--- a/src/gateway/events/Connection.ts
+++ b/src/gateway/events/Connection.ts
@@ -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);
}
}
diff --git a/src/gateway/events/Message.ts b/src/gateway/events/Message.ts
index 7a2e49d3..d143ee69 100644
--- a/src/gateway/events/Message.ts
+++ b/src/gateway/events/Message.ts
@@ -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);
}
}
diff --git a/src/gateway/listener/listener.ts b/src/gateway/listener/listener.ts
index b2b7fbf6..fca34b13 100644
--- a/src/gateway/listener/listener.ts
+++ b/src/gateway/listener/listener.ts
@@ -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
diff --git a/src/gateway/opcodes/Identify.ts b/src/gateway/opcodes/Identify.ts
index 48fd3a20..48e24afa 100644
--- a/src/gateway/opcodes/Identify.ts
+++ b/src/gateway/opcodes/Identify.ts
@@ -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();
diff --git a/src/gateway/opcodes/StreamCreate.ts b/src/gateway/opcodes/StreamCreate.ts
index 6fc222ab..b1695250 100644
--- a/src/gateway/opcodes/StreamCreate.ts
+++ b/src/gateway/opcodes/StreamCreate.ts
@@ -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;
diff --git a/src/gateway/opcodes/StreamDelete.ts b/src/gateway/opcodes/StreamDelete.ts
index 82d09538..773b6508 100644
--- a/src/gateway/opcodes/StreamDelete.ts
+++ b/src/gateway/opcodes/StreamDelete.ts
@@ -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?
diff --git a/src/gateway/opcodes/StreamWatch.ts b/src/gateway/opcodes/StreamWatch.ts
index f63e4a15..5237e766 100644
--- a/src/gateway/opcodes/StreamWatch.ts
+++ b/src/gateway/opcodes/StreamWatch.ts
@@ -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,
diff --git a/src/gateway/opcodes/instanceOf.ts b/src/gateway/opcodes/instanceOf.ts
index 7bebf49e..683feeb5 100644
--- a/src/gateway/opcodes/instanceOf.ts
+++ b/src/gateway/opcodes/instanceOf.ts
@@ -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;
}
}
diff --git a/src/gateway/routes/_spacebar/gateway/admin/introspect.ts b/src/gateway/routes/_spacebar/gateway/admin/introspect.ts
index f6ab1e7b..d34e17cf 100644
--- a/src/gateway/routes/_spacebar/gateway/admin/introspect.ts
+++ b/src/gateway/routes/_spacebar/gateway/admin/introspect.ts
@@ -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,
diff --git a/src/gateway/util/Heartbeat.ts b/src/gateway/util/Heartbeat.ts
index 5beb7e99..6ec067cb 100644
--- a/src/gateway/util/Heartbeat.ts
+++ b/src/gateway/util/Heartbeat.ts
@@ -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);
}
diff --git a/src/gateway/util/Send.ts b/src/gateway/util/Send.ts
index f857a400..60eda0e2 100644
--- a/src/gateway/util/Send.ts
+++ b/src/gateway/util/Send.ts
@@ -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);
});
diff --git a/src/gateway/util/WebSocket.ts b/src/gateway/util/WebSocket.ts
index 50c7bbb1..cf912ddf 100644
--- a/src/gateway/util/WebSocket.ts
+++ b/src/gateway/util/WebSocket.ts
@@ -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;
diff --git a/src/webrtc/events/Close.ts b/src/webrtc/events/Close.ts
index e88b2938..bf2b4e22 100644
--- a/src/webrtc/events/Close.ts
+++ b/src/webrtc/events/Close.ts
@@ -18,8 +18,8 @@
import { WebSocket } from "@spacebar/gateway";
-export async function onClose(this: WebSocket, code: number, reason: string) {
+export async function onClose(socket: WebSocket, code: number, reason: Buffer) {
console.log("[WebRTC] closed", code, reason.toString());
- this.removeAllListeners();
+ socket.rawSocket.removeAllListeners();
}
diff --git a/src/webrtc/events/Connection.ts b/src/webrtc/events/Connection.ts
index f1808399..2949ea75 100644
--- a/src/webrtc/events/Connection.ts
+++ b/src/webrtc/events/Connection.ts
@@ -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")) || 5;
- 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);
}
}
diff --git a/src/webrtc/events/Message.ts b/src/webrtc/events/Message.ts
index b5455d97..0ec2108a 100644
--- a/src/webrtc/events/Message.ts
+++ b/src/webrtc/events/Message.ts
@@ -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);
diff --git a/src/webrtc/opcodes/Heartbeat.ts b/src/webrtc/opcodes/Heartbeat.ts
index ee0e72b3..8e38a11a 100644
--- a/src/webrtc/opcodes/Heartbeat.ts
+++ b/src/webrtc/opcodes/Heartbeat.ts
@@ -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 });
}
diff --git a/src/webrtc/opcodes/Identify.ts b/src/webrtc/opcodes/Identify.ts
index 7d776d9c..fdf5f449 100644
--- a/src/webrtc/opcodes/Identify.ts
+++ b/src/webrtc/opcodes/Identify.ts
@@ -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!);
});
diff --git a/src/webrtc/opcodes/SelectProtocol.ts b/src/webrtc/opcodes/SelectProtocol.ts
index 6a733932..c3b61554 100644
--- a/src/webrtc/opcodes/SelectProtocol.ts
+++ b/src/webrtc/opcodes/SelectProtocol.ts
@@ -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 { VoiceOPCodes, VoicePayload, WebRtcWebSocket, mediaServer, Send } 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 ?? []);
diff --git a/src/webrtc/util/Send.ts b/src/webrtc/util/Send.ts
index 1d8aa61e..c3f47bae 100644
--- a/src/webrtc/util/Send.ts
+++ b/src/webrtc/util/Send.ts
@@ -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);
});
diff --git a/src/webrtc/util/WebRtcWebSocket.ts b/src/webrtc/util/WebRtcWebSocket.ts
index e36b4f1d..8d07dfdf 100644
--- a/src/webrtc/util/WebRtcWebSocket.ts
+++ b/src/webrtc/util/WebRtcWebSocket.ts
@@ -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>;
}
|
