diff --git a/pkgs/gateway/package.json b/pkgs/gateway/package.json index 722e805..6a40351 100644 --- a/pkgs/gateway/package.json +++ b/pkgs/gateway/package.json @@ -1,6 +1,6 @@ { "name": "@purrkit/gateway", - "version": "1.2.0", + "version": "1.3.0", "license": "EUPL-1.2", "files": [ "dist", diff --git a/pkgs/gateway/src/client.ts b/pkgs/gateway/src/client.ts index e05048e..62528be 100644 --- a/pkgs/gateway/src/client.ts +++ b/pkgs/gateway/src/client.ts @@ -1,47 +1,107 @@ import { EventEmitter } from "events"; -import { createInflate, type Inflate } from "zlib"; +import { createInflate, constants, type Inflate } from "zlib"; import { OpCodes, type GatewayEvents, type GatewayState } from "./types"; -const ZLIB_SUFFIX = new Uint8Array([0x00, 0x00, 0xff, 0xff]); - -function hasZSyncFlush(data: Uint8Array): boolean { - if (data.length < 4) return false; - const tail = data.subarray(data.length - 4); - return ( - tail[0] === ZLIB_SUFFIX[0] && - tail[1] === ZLIB_SUFFIX[1] && - tail[2] === ZLIB_SUFFIX[2] && - tail[3] === ZLIB_SUFFIX[3] - ); +const ZLIB_FLUSH_MARKER: ReadonlyArray = [0x00, 0x00, 0xff, 0xff]; + +function indexOfZSyncFlush(data: Buffer): number { + for (let i = 0; i + 4 <= data.length; i++) { + if ( + data[i] === ZLIB_FLUSH_MARKER[0] && + data[i + 1] === ZLIB_FLUSH_MARKER[1] && + data[i + 2] === ZLIB_FLUSH_MARKER[2] && + data[i + 3] === ZLIB_FLUSH_MARKER[3] + ) { + return i; + } + } + return -1; +} + +function splitJsonObjects(text: string): { messages: string[]; leftover: string } { + const messages: string[] = []; + let depth = 0; + let inString = false; + let escaped = false; + let start = -1; + + for (let i = 0; i < text.length; i++) { + const ch = text[i]; + + if (inString) { + if (escaped) { + escaped = false; + } else if (ch === "\\") { + escaped = true; + } else if (ch === '"') { + inString = false; + } + continue; + } + + if (ch === '"') { + inString = true; + } else if (ch === "{") { + if (depth === 0) start = i; + depth++; + } else if (ch === "}") { + depth--; + if (depth === 0 && start !== -1) { + messages.push(text.slice(start, i + 1)); + } + } + } + + if (depth > 0 && start !== -1) { + return { messages, leftover: text.slice(start) }; + } + + return { messages, leftover: "" }; } class ZlibStream { private inflater: Inflate; - private buffer: Uint8Array[]; + private pending: Buffer; + private leftover: string; constructor() { - this.buffer = []; - this.inflater = createInflate(); + this.pending = Buffer.alloc(0); + this.leftover = ""; + this.inflater = createInflate({ flush: constants.Z_SYNC_FLUSH }); } - async push(data: Uint8Array): Promise { - this.buffer.push(data); - if (!hasZSyncFlush(data)) return null; + async push(data: Uint8Array): Promise { + this.pending = Buffer.concat([ + this.pending, + Buffer.from(data.buffer, data.byteOffset, data.byteLength), + ]); + + const messages: string[] = []; + + for (;;) { + const idx = indexOfZSyncFlush(this.pending); + if (idx === -1) break; + + const segment = Buffer.from(this.pending.subarray(0, idx + 4)); + this.pending = this.pending.subarray(idx + 4); + + const decoded = this.leftover + (await this.inflate(segment)); + this.leftover = ""; - const totalLen = this.buffer.reduce((sum, c) => sum + c.length, 0); - const full = new Uint8Array(totalLen); - let offset = 0; - for (const chunk of this.buffer) { - full.set(chunk, offset); - offset += chunk.length; + const { messages: split, leftover } = splitJsonObjects(decoded); + this.leftover = leftover; + messages.push(...split); } - this.buffer = []; + return messages; + } + + private inflate(segment: Buffer): Promise { return new Promise((resolve) => { const chunks: Buffer[] = []; const onData = (chunk: Buffer) => chunks.push(chunk); this.inflater.on("data", onData); - this.inflater.write(Buffer.from(full.buffer, full.byteOffset, full.byteLength), () => { + this.inflater.write(segment, () => { this.inflater.removeListener("data", onData); resolve(Buffer.concat(chunks).toString("utf-8")); }); @@ -49,7 +109,8 @@ class ZlibStream { } end() { - this.buffer = []; + this.pending = Buffer.alloc(0); + this.leftover = ""; } } @@ -262,11 +323,9 @@ export function Gateway(options: GatewayClientOptions): GatewayClientInstance { ws.onopen = () => {}; ws.onmessage = async (event) => { - let payload: { op: OpCodes; d: unknown; s: number | null; t: string | null }; + let raws: string[]; try { - let rawData: string; - if (options.compress && zlibStream) { let bytes: Uint8Array; if (typeof event.data === "string") { @@ -279,80 +338,95 @@ export function Gateway(options: GatewayClientOptions): GatewayClientInstance { bytes = new Uint8Array(event.data as ArrayBuffer); } - const result = await zlibStream.push(bytes); - if (result === null) return; - - rawData = result; + const messages = await zlibStream.push(bytes); + if (messages.length === 0) return; + raws = messages; } else { - rawData = + raws = [ typeof event.data === "string" ? event.data - : new TextDecoder().decode(event.data as ArrayBuffer); + : new TextDecoder().decode(event.data as ArrayBuffer), + ]; } - - payload = JSON.parse(rawData); } catch (e) { emitter.emit("error", `Failed to parse gateway payload: ${String(e)}`); + emitter.emit("error", event); return; } - const { op, d, s, t } = payload; + for (const rawData of raws) { + let payload: { op: OpCodes; d: unknown; s: number | null; t: string | null }; - if (s !== null && s !== undefined) { - sequenceNum = s; - } + emitter.emit("info", rawData); + try { + payload = JSON.parse(rawData); + } catch (e) { + emitter.emit("error", `Failed to parse gateway payload: ${String(e)}`); + emitter.emit("error", rawData); + continue; + } - switch (op) { - case OpCodes.Dispatch: - if (t === "READY") { - const readyData = d as { session_id: string; resume_gateway_url: string }; - sessionId = readyData.session_id; - resumeGatewayUrl = readyData.resume_gateway_url; - state = "ready"; - reconnectAttempts = 0; - readyResolve?.(); - emitter.emit("info", "Gateway ready"); + const { op, d, s, t } = payload; + + if (s !== null && s !== undefined) { + sequenceNum = s; + } + + switch (op) { + case OpCodes.Dispatch: + if (t === "READY") { + const readyData = d as { + session_id: string; + resume_gateway_url: string; + }; + sessionId = readyData.session_id; + resumeGatewayUrl = readyData.resume_gateway_url; + state = "ready"; + reconnectAttempts = 0; + readyResolve?.(); + emitter.emit("info", "Gateway ready"); + } + if (t === "RESUMED") { + state = "ready"; + reconnectAttempts = 0; + readyResolve?.(); + } + emitter.emit(t as string, d); + break; + case OpCodes.Heartbeat: + send(OpCodes.Heartbeat, sequenceNum); + break; + case OpCodes.Reconnect: + emitter.emit("info", "Gateway requested reconnect"); + handleReconnect(); + break; + case OpCodes.InvalidSession: { + const jitterDelay = Math.floor(Math.random() * 4000) + 1000; + if (d as boolean) { + handleReconnect(jitterDelay); + } else { + sequenceNum = null; + sessionId = null; + resumeGatewayUrl = null; + handleReconnect(jitterDelay); + } + break; } - if (t === "RESUMED") { - state = "ready"; + case OpCodes.Hello: { + stopConnectionTimeout(); + startHeartbeat((d as { heartbeat_interval: number }).heartbeat_interval); reconnectAttempts = 0; - readyResolve?.(); - } - emitter.emit(t as string, d); - break; - case OpCodes.Heartbeat: - send(OpCodes.Heartbeat, sequenceNum); - break; - case OpCodes.Reconnect: - emitter.emit("info", "Gateway requested reconnect"); - handleReconnect(); - break; - case OpCodes.InvalidSession: { - const jitterDelay = Math.floor(Math.random() * 4000) + 1000; - if (d as boolean) { - handleReconnect(jitterDelay); - } else { - sequenceNum = null; - sessionId = null; - resumeGatewayUrl = null; - handleReconnect(jitterDelay); - } - break; - } - case OpCodes.Hello: { - stopConnectionTimeout(); - startHeartbeat((d as { heartbeat_interval: number }).heartbeat_interval); - reconnectAttempts = 0; - if (sessionId && resumeGatewayUrl) { - resume(); - } else { - identify(); + if (sessionId && resumeGatewayUrl) { + resume(); + } else { + identify(); + } + break; } - break; + case OpCodes.HeartbeatAck: + isHeartbeatAcked = true; + break; } - case OpCodes.HeartbeatAck: - isHeartbeatAcked = true; - break; } }; diff --git a/pkgs/gateway/src/types.ts b/pkgs/gateway/src/types.ts index adcce90..a9ab5cd 100644 --- a/pkgs/gateway/src/types.ts +++ b/pkgs/gateway/src/types.ts @@ -398,7 +398,9 @@ export interface GatewayInviteDeleteDispatchData { code: string; } -export type GatewayMessageCreateDispatchData = MessageResponse; +export type GatewayMessageCreateDispatchData = MessageResponse & { + member?: GuildMemberResponse; +}; export interface GatewayMessageUpdateDispatchData extends Partial { id: Snowflake;