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);
}
}
|
