diff --git a/mod.ts b/mod.ts index 07db84e82..f18372487 100644 --- a/mod.ts +++ b/mod.ts @@ -1,5 +1,3 @@ -import createClient from "./src/module/client.ts"; - export * from "./src/controllers/bans.ts"; export * from "./src/controllers/cache.ts"; export * from "./src/controllers/channels.ts"; @@ -42,5 +40,3 @@ export * from "./src/utils/cdn.ts"; export * from "./src/utils/collection.ts"; export * from "./src/utils/permissions.ts"; export * from "./src/utils/utils.ts"; - -export default createClient; diff --git a/src/module/basicShard.ts b/src/module/basicShard.ts deleted file mode 100644 index ee1a9cc04..000000000 --- a/src/module/basicShard.ts +++ /dev/null @@ -1,404 +0,0 @@ -import { - connectWebSocket, - delay, - inflate, - isWebSocketCloseEvent, - isWebSocketPingEvent, - isWebSocketPongEvent, - WebSocket, -} from "../../deps.ts"; -import { - DiscordBotGatewayData, - DiscordHeartbeatPayload, - GatewayOpcode, - ReadyPayload, -} from "../types/discord.ts"; -import { FetchMembersOptions } from "../types/guild.ts"; -import { BotStatusRequest } from "../utils/utils.ts"; -import { botGatewayData, eventHandlers, IdentifyPayload } from "./client.ts"; -import { handleDiscordPayload } from "./shardingManager.ts"; - -const basicShards = new Map(); -const heartbeating = new Map(); -const utf8decoder = new TextDecoder(); -const RequestMembersQueue: RequestMemberQueuedRequest[] = []; -let processQueue = false; - -export interface BasicShard { - id: number; - socket: WebSocket; - resumeInterval: number; - sessionID: string; - previousSequenceNumber: number | null; - needToResume: boolean; -} - -interface RequestMemberQueuedRequest { - guildID: string; - shardID: number; - nonce: string; - options?: FetchMembersOptions; -} - -export async function createBasicShard( - data: DiscordBotGatewayData, - identifyPayload: IdentifyPayload, - resuming = false, - shardID = 0, -) { - const oldShard = basicShards.get(shardID); - - const basicShard: BasicShard = { - id: shardID, - socket: await connectWebSocket(`${data.url}?v=8&encoding=json`), - resumeInterval: 0, - sessionID: oldShard?.sessionID || "", - previousSequenceNumber: oldShard?.previousSequenceNumber || 0, - needToResume: false, - }; - - basicShards.set(basicShard.id, basicShard); - - if (!resuming) { - // Intial identify with the gateway - await identify(basicShard, identifyPayload); - } else { - await resume(basicShard, identifyPayload); - } - - for await (let message of basicShard.socket) { - if (isWebSocketCloseEvent(message)) { - eventHandlers.debug?.( - { type: "websocketClose", data: { shardID: basicShard.id, message } }, - ); - - // These error codes should just crash the projects - if ([4004, 4005, 4012, 4013, 4014].includes(message.code)) { - console.error(`Close :( ${JSON.stringify(message)}`); - eventHandlers.debug?.( - { - type: "websocketErrored", - data: { shardID: basicShard.id, message }, - }, - ); - - throw new Error( - "Shard.ts: Error occurred that is not resumeable or able to be reconnected.", - ); - } - // These error codes can not be resumed but need to reconnect from start - if ([4003, 4007, 4008, 4009].includes(message.code)) { - eventHandlers.debug?.( - { - type: "websocketReconnecting", - data: { shardID: basicShard.id, message }, - }, - ); - createBasicShard(botGatewayData, identifyPayload, false, shardID); - } else { - basicShard.needToResume = true; - resumeConnection(botGatewayData, identifyPayload, basicShard.id); - } - continue; - } else if (isWebSocketPingEvent(message) || isWebSocketPongEvent(message)) { - continue; - } - - if (message instanceof Uint8Array) { - message = inflate( - message, - 0, - (slice: Uint8Array) => utf8decoder.decode(slice), - ); - } - - if (typeof message === "string") { - const data = JSON.parse(message); - if (!data.t) eventHandlers.rawGateway?.(data); - switch (data.op) { - case GatewayOpcode.Hello: - if (!heartbeating.has(basicShard.id)) { - heartbeat( - basicShard, - (data.d as DiscordHeartbeatPayload).heartbeat_interval, - identifyPayload, - ); - } - break; - case GatewayOpcode.HeartbeatACK: - heartbeating.set(shardID, true); - break; - case GatewayOpcode.Reconnect: - eventHandlers.debug?.( - { type: "reconnect", data: { shardID: basicShard.id } }, - ); - basicShard.needToResume = true; - resumeConnection(botGatewayData, identifyPayload, basicShard.id); - break; - case GatewayOpcode.InvalidSession: - eventHandlers.debug?.( - { type: "invalidSession", data: { shardID: basicShard.id, data } }, - ); - // When d is false we need to reidentify - if (!data.d) { - createBasicShard(botGatewayData, identifyPayload, false, shardID); - break; - } - basicShard.needToResume = true; - resumeConnection(botGatewayData, identifyPayload, basicShard.id); - break; - default: - if (data.t === "RESUMED") { - eventHandlers.debug?.( - { type: "resumed", data: { shardID: basicShard.id } }, - ); - - basicShard.needToResume = false; - break; - } - // Important for RESUME - if (data.t === "READY") { - basicShard.sessionID = (data.d as ReadyPayload).session_id; - } - - // Update the sequence number if it is present - if (data.s) basicShard.previousSequenceNumber = data.s; - - handleDiscordPayload(data, basicShard.id); - break; - } - } - } -} - -function identify(shard: BasicShard, payload: IdentifyPayload) { - eventHandlers.debug?.( - { - type: "identifying", - data: { - shardID: shard.id, - }, - }, - ); - - return shard.socket.send( - JSON.stringify( - { - op: GatewayOpcode.Identify, - d: { ...payload, shard: [shard.id, payload.shard[1]] }, - }, - ), - ); -} - -function resume(shard: BasicShard, payload: IdentifyPayload) { - return shard.socket.send(JSON.stringify({ - op: GatewayOpcode.Resume, - d: { - token: payload.token, - session_id: shard.sessionID, - seq: shard.previousSequenceNumber, - }, - })); -} - -async function heartbeat( - shard: BasicShard, - interval: number, - payload: IdentifyPayload, -) { - // We lost socket connection between heartbeats, resume connection - if (shard.socket.isClosed) { - shard.needToResume = true; - resumeConnection(botGatewayData, payload, shard.id); - heartbeating.delete(shard.id); - return; - } - - if (heartbeating.has(shard.id)) { - const receivedACK = heartbeating.get(shard.id); - // If a ACK response was not received since last heartbeat, issue invalid session close - if (!receivedACK) { - eventHandlers.debug?.( - { - type: "heartbeatStopped", - data: { - interval, - previousSequenceNumber: shard.previousSequenceNumber, - shardID: shard.id, - }, - }, - ); - return shard.socket.send(JSON.stringify({ op: 4009 })); - } - } - - // Set it to false as we are issuing a new heartbeat - heartbeating.set(shard.id, false); - - shard.socket.send( - JSON.stringify( - { op: GatewayOpcode.Heartbeat, d: shard.previousSequenceNumber }, - ), - ); - eventHandlers.debug?.( - { - type: "heartbeat", - data: { - interval, - previousSequenceNumber: shard.previousSequenceNumber, - shardID: shard.id, - }, - }, - ); - await delay(interval); - heartbeat(shard, interval, payload); -} - -async function resumeConnection( - botGatewayData: DiscordBotGatewayData, - payload: IdentifyPayload, - shardID: number, -) { - const shard = basicShards.get(shardID); - if (!shard) { - eventHandlers.debug?.( - { type: "missingShard", data: { shardID: shardID } }, - ); - return; - } - - if (!shard.needToResume) return; - - eventHandlers.debug?.({ type: "resuming", data: { shardID: shard.id } }); - // Run it once - createBasicShard(botGatewayData, payload, true, shard.id); - // Then retry every 15 seconds - await delay(1000 * 15); - if (shard.needToResume) resumeConnection(botGatewayData, payload, shardID); -} - -export function requestGuildMembers( - guildID: string, - shardID: number, - nonce: string, - options?: FetchMembersOptions, - queuedRequest = false, -) { - const shard = basicShards.get(shardID); - - // This request was not from this queue so we add it to queue first - if (!queuedRequest) { - RequestMembersQueue.push({ - guildID, - shardID, - nonce, - options, - }); - - if (!processQueue) { - processQueue = true; - processGatewayQueue(); - } - return; - } - - // If its closed add back to queue to redo on resume - if (shard?.socket.isClosed) { - requestGuildMembers(guildID, shardID, nonce, options); - return; - } - - shard?.socket.send(JSON.stringify({ - op: GatewayOpcode.RequestGuildMembers, - d: { - guild_id: guildID, - query: options?.query || "", - limit: options?.limit || 0, - presences: options?.presences || false, - user_ids: options?.userIDs, - nonce, - }, - })); -} - -async function processGatewayQueue() { - if (!RequestMembersQueue.length) { - processQueue = false; - return; - } - - basicShards.forEach((shard) => { - const index = RequestMembersQueue.findIndex((q) => q.shardID === shard.id); - // 2 events per second is the rate limit. - const request = RequestMembersQueue[index]; - if (request) { - eventHandlers.debug?.( - { - type: "requestMembersProcessing", - data: { - remaining: RequestMembersQueue.length, - request, - }, - }, - ); - requestGuildMembers( - request.guildID, - request.shardID, - request.nonce, - request.options, - true, - ); - // Remove item from queue - RequestMembersQueue.splice(index, 1); - - const secondIndex = RequestMembersQueue.findIndex((q) => - q.shardID === shard.id - ); - const secondRequest = RequestMembersQueue[secondIndex]; - if (secondRequest) { - eventHandlers.debug?.( - { - type: "requestMembersProcessing", - data: { - remaining: RequestMembersQueue.length, - request, - }, - }, - ); - requestGuildMembers( - secondRequest.guildID, - secondRequest.shardID, - secondRequest.nonce, - secondRequest.options, - true, - ); - // Remove item from queue - RequestMembersQueue.splice(secondIndex, 1); - } - } - }); - - await delay(1500); - - processGatewayQueue(); -} - -export function botGatewayStatusRequest(payload: BotStatusRequest) { - basicShards.forEach((shard) => { - shard.socket.send(JSON.stringify({ - op: GatewayOpcode.StatusUpdate, - d: { - since: null, - game: payload.game.name - ? { - name: payload.game.name, - type: payload.game.type, - } - : null, - status: payload.status, - afk: false, - }, - })); - }); -} diff --git a/src/module/client.ts b/src/module/client.ts index 6b0104da3..5172813e4 100644 --- a/src/module/client.ts +++ b/src/module/client.ts @@ -1,6 +1,6 @@ import { DiscordBotGatewayData } from "../types/discord.ts"; import { ClientOptions, EventHandlers } from "../types/options.ts"; -import { endpoints } from "../utils/constants.ts"; +import { baseEndpoints, endpoints } from "../utils/constants.ts"; import { RequestManager } from "./requestManager.ts"; import { spawnShards } from "./shardingManager.ts"; @@ -10,6 +10,7 @@ export let botID = ""; export let eventHandlers: EventHandlers = {}; export let botGatewayData: DiscordBotGatewayData; +export let proxyWSURL = ""; export const identifyPayload: IdentifyPayload = { token: "", @@ -35,7 +36,7 @@ export interface IdentifyPayload { shard: [number, number]; } -export const createClient = async (data: ClientOptions) => { +export async function createClient(data: ClientOptions) { if (data.eventHandlers) eventHandlers = data.eventHandlers; authorization = `Bot ${data.token}`; @@ -51,8 +52,8 @@ export const createClient = async (data: ClientOptions) => { ); identifyPayload.shard = [0, botGatewayData.shards]; - spawnShards(botGatewayData, identifyPayload); -}; + spawnShards(botGatewayData, identifyPayload, 0, botGatewayData.shards); +} export default createClient; @@ -63,3 +64,56 @@ export function updateEventHandlers(newEventHandlers: EventHandlers) { export function setBotID(id: string) { if (botID !== id) botID = id; } + +// BIG BRAIN BOT STUFF ONLY BELOW THIS + +/** + * This function should be used only by bot developers whose bots are in over 25,000 servers. + * Please be aware if you are a beginner developer using this, things will not work as per the guides. This is for advanced developers only! + * + * Advanced Devs: This function will allow you to have an insane amount of customization potential as when you get to large bots you need to be able to optimize every tiny detail to make you bot work the way you need. +*/ +export async function startBigBrainBot(data: BigBrainBotOptions) { + authorization = `Bot ${data.token}`; + identifyPayload.token = `Bot ${data.token}`; + + if (data.restURL) baseEndpoints.BASE_URL = data.restURL; + if (data.cdnURL) baseEndpoints.CDN_URL = data.cdnURL; + if (data.wsURL) proxyWSURL = data.wsURL; + if (data.eventHandlers) eventHandlers = data.eventHandlers; + if (data.compress) { + identifyPayload.compress = data.compress; + } + + identifyPayload.intents = data.intents.reduce( + (bits, next) => (bits |= next), + 0, + ); + + // Initial API connection to get info about bots connection + botGatewayData = await RequestManager.get( + endpoints.GATEWAY_BOT, + ) as DiscordBotGatewayData; + + spawnShards( + botGatewayData, + identifyPayload, + data.firstShardID, + data.lastShardID || botGatewayData.shards >= 25 + ? (data.firstShardID + 25) + : botGatewayData.shards, + ); +} + +export interface BigBrainBotOptions extends ClientOptions { + /** The first shard to start at for this worker. Use this to control which shards to run in each worker. */ + firstShardID: number; + /** The last shard to start for this worker. By default it will be 25 + the firstShardID. */ + lastShardID?: number; + /** This can be used to forward the ws handling to a proxy. */ + wsURL?: string; + /** This can be used to forward the REST handling to a proxy. */ + restURL?: string; + /** This can be used to forward the CDN handling to a proxy. */ + cdnURL?: string; +} diff --git a/src/module/requestManager.ts b/src/module/requestManager.ts index 80a53fb54..bdec45be9 100644 --- a/src/module/requestManager.ts +++ b/src/module/requestManager.ts @@ -2,7 +2,7 @@ import { delay } from "../../deps.ts"; import { HttpResponseCode } from "../types/discord.ts"; import { Errors } from "../types/errors.ts"; import { RequestMethods } from "../types/fetch.ts"; -import { baseEndpoints } from "../utils/constants.ts"; +import { baseEndpoints, discordAPIURLS } from "../utils/constants.ts"; import { authorization, eventHandlers } from "./client.ts"; const pathQueues: { [key: string]: QueuedRequest[] } = {}; @@ -144,7 +144,7 @@ function createRequestBody(body: any, method: RequestMethods) { const headers: { [key: string]: string } = { Authorization: authorization, "User-Agent": - `DiscordBot (https://github.com/skillz4killz/discordeno, 6.0.0)`, + `DiscordBot (https://github.com/skillz4killz/discordeno, v10)`, }; if (method === "get") body = undefined; @@ -203,6 +203,20 @@ async function runMethod( const errorStack = new Error("Location:"); Error.captureStackTrace(errorStack); + // For proxies we don't need to do any of the legwork so we just forward the request + if ( + !url.startsWith(discordAPIURLS.BASE_URL) && + !url.startsWith(discordAPIURLS.CDN_URL) + ) { + return fetch(url, { method, body: body ? JSON.stringify(body) : undefined }) + .then((res) => res.json()) + .catch((error) => { + console.error(error); + throw errorStack; + }); + } + + // No proxy so we need to handl all rate limiting and such return new Promise((resolve, reject) => { const callback = async () => { try { diff --git a/src/module/shard.ts b/src/module/shard.ts index 8e7353127..b6ce065cd 100644 --- a/src/module/shard.ts +++ b/src/module/shard.ts @@ -1,9 +1,13 @@ import { connectWebSocket, delay, + inflate, isWebSocketCloseEvent, + isWebSocketPingEvent, + isWebSocketPongEvent, WebSocket, } from "../../deps.ts"; +import { eventHandlers } from "../../mod.ts"; import { DiscordBotGatewayData, DiscordHeartbeatPayload, @@ -11,167 +15,75 @@ import { ReadyPayload, } from "../types/discord.ts"; import { FetchMembersOptions } from "../types/guild.ts"; -import { DebugArg } from "../types/options.ts"; - -let shardSocket: WebSocket; - -/** The session id is needed for RESUME functionality when discord disconnects randomly. */ -let sessionID = ""; - -// Discord requests null if no number has yet been sent by discord -let previousSequenceNumber: number | null = null; -let needToResume = false; -let shardID = 0; +import { BotStatusRequest } from "../utils/utils.ts"; +import { IdentifyPayload, proxyWSURL } from "./client.ts"; +import { handleDiscordPayload } from "./shardingManager.ts"; +const basicShards = new Map(); +const heartbeating = new Map(); +const utf8decoder = new TextDecoder(); const RequestMembersQueue: RequestMemberQueuedRequest[] = []; let processQueue = false; +export interface BasicShard { + id: number; + socket: WebSocket; + resumeInterval: number; + sessionID: string; + previousSequenceNumber: number | null; + needToResume: boolean; +} + interface RequestMemberQueuedRequest { guildID: string; + shardID: number; nonce: string; options?: FetchMembersOptions; } -async function processRequestMembersQueue() { - if (!RequestMembersQueue.length) { - processQueue = false; - return; - } - - // 2 events per second is the rate limit. - const request = RequestMembersQueue.shift(); - if (request) { - requestGuildMembers(request.guildID, request.nonce, request.options, true); - - const secondRequest = RequestMembersQueue.shift(); - if (secondRequest) { - requestGuildMembers( - secondRequest.guildID, - secondRequest.nonce, - secondRequest.options, - true, - ); - } - } - - await delay(1500); - - postDebug( - { - type: "requestMembersProcessing", - data: { shardID, remaining: RequestMembersQueue.length }, - }, - ); - processRequestMembersQueue(); -} - -// TODO: If a client does not receive a heartbeat ack between its attempts at sending heartbeats, it should immediately terminate the connection with a non-1000 close code, reconnect, and attempt to resume. -async function sendConstantHeartbeats( - interval: number, -) { - await delay(interval); - shardSocket.send( - JSON.stringify({ op: GatewayOpcode.Heartbeat, d: previousSequenceNumber }), - ); - postDebug( - { type: "heartbeat", data: { interval, previousSequenceNumber, shardID } }, - ); - - sendConstantHeartbeats(interval); -} - -async function resumeConnection( - botGatewayData: DiscordBotGatewayData, - identifyPayload: object, -) { - postDebug({ type: "resuming", data: { shardID } }); - // Run it once - createShard(botGatewayData, identifyPayload, true); - // Then retry every 15 seconds - await delay(1000 * 15); - if (needToResume) resumeConnection(botGatewayData, identifyPayload); -} - -const createShard = async ( - botGatewayData: DiscordBotGatewayData, - identifyPayload: object, +export async function createShard( + data: DiscordBotGatewayData, + identifyPayload: IdentifyPayload, resuming = false, -) => { - postDebug({ type: "createShard", data: { shardID } }); + shardID = 0, +) { + const oldShard = basicShards.get(shardID); - shardSocket = await connectWebSocket(botGatewayData.url); - let resumeInterval = 0; + const basicShard: BasicShard = { + id: shardID, + socket: await connectWebSocket( + proxyWSURL || `${data.url}?v=8&encoding=json`, + ), + resumeInterval: 0, + sessionID: oldShard?.sessionID || "", + previousSequenceNumber: oldShard?.previousSequenceNumber || 0, + needToResume: false, + }; + + basicShards.set(basicShard.id, basicShard); if (!resuming) { // Intial identify with the gateway - await shardSocket.send( - JSON.stringify({ op: GatewayOpcode.Identify, d: identifyPayload }), - ); + await identify(basicShard, identifyPayload); } else { - await shardSocket.send(JSON.stringify({ - op: GatewayOpcode.Resume, - d: { - ...identifyPayload, - session_id: sessionID, - seq: previousSequenceNumber, - }, - })); + await resume(basicShard, identifyPayload); } - for await (const message of shardSocket) { - if (typeof message === "string") { - const data = JSON.parse(message); - - switch (data.op) { - case GatewayOpcode.Hello: - sendConstantHeartbeats( - (data.d as DiscordHeartbeatPayload).heartbeat_interval, - ); - break; - case GatewayOpcode.Reconnect: - case GatewayOpcode.InvalidSession: - // When d is false we need to reidentify - if (!data.d) { - postDebug({ type: "invalidSession", data: { shardID } }); - createShard(botGatewayData, identifyPayload); - break; - } - needToResume = true; - resumeConnection(botGatewayData, identifyPayload); - break; - default: - if (data.t === "RESUMED") { - postDebug({ type: "resumed", data: { shardID } }); - - needToResume = false; - break; - } - // Important for RESUME - if (data.t === "READY") { - sessionID = (data.d as ReadyPayload).session_id; - } - - // Update the sequence number if it is present - if (data.s) previousSequenceNumber = data.s; - - // @ts-ignore - postMessage( - { - type: "HANDLE_DISCORD_PAYLOAD", - payload: message, - resumeInterval, - shardID, - }, - ); - break; - } - } else if (isWebSocketCloseEvent(message)) { - postDebug({ type: "websocketClose", data: { shardID, message } }); + for await (let message of basicShard.socket) { + if (isWebSocketCloseEvent(message)) { + eventHandlers.debug?.( + { type: "websocketClose", data: { shardID: basicShard.id, message } }, + ); // These error codes should just crash the projects if ([4004, 4005, 4012, 4013, 4014].includes(message.code)) { console.error(`Close :( ${JSON.stringify(message)}`); - postDebug({ type: "websocketErrored", data: { shardID, message } }); + eventHandlers.debug?.( + { + type: "websocketErrored", + data: { shardID: basicShard.id, message }, + }, + ); throw new Error( "Shard.ts: Error occurred that is not resumeable or able to be reconnected.", @@ -179,51 +91,235 @@ const createShard = async ( } // These error codes can not be resumed but need to reconnect from start if ([4003, 4007, 4008, 4009].includes(message.code)) { - postDebug( - { type: "websocketReconnecting", data: { shardID, message } }, + eventHandlers.debug?.( + { + type: "websocketReconnecting", + data: { shardID: basicShard.id, message }, + }, ); - createShard(botGatewayData, identifyPayload); + createShard(data, identifyPayload, false, shardID); } else { - needToResume = true; - resumeConnection(botGatewayData, identifyPayload); + basicShard.needToResume = true; + resumeConnection(data, identifyPayload, basicShard.id); + } + continue; + } else if (isWebSocketPingEvent(message) || isWebSocketPongEvent(message)) { + continue; + } + + if (message instanceof Uint8Array) { + message = inflate( + message, + 0, + (slice: Uint8Array) => utf8decoder.decode(slice), + ); + } + + if (typeof message === "string") { + const data = JSON.parse(message); + if (!data.t) eventHandlers.rawGateway?.(data); + switch (data.op) { + case GatewayOpcode.Hello: + if (!heartbeating.has(basicShard.id)) { + heartbeat( + basicShard, + (data.d as DiscordHeartbeatPayload).heartbeat_interval, + identifyPayload, + data, + ); + } + break; + case GatewayOpcode.HeartbeatACK: + heartbeating.set(shardID, true); + break; + case GatewayOpcode.Reconnect: + eventHandlers.debug?.( + { type: "reconnect", data: { shardID: basicShard.id } }, + ); + basicShard.needToResume = true; + resumeConnection(data, identifyPayload, basicShard.id); + break; + case GatewayOpcode.InvalidSession: + eventHandlers.debug?.( + { type: "invalidSession", data: { shardID: basicShard.id, data } }, + ); + // When d is false we need to reidentify + if (!data.d) { + createShard(data, identifyPayload, false, shardID); + break; + } + basicShard.needToResume = true; + resumeConnection(data, identifyPayload, basicShard.id); + break; + default: + if (data.t === "RESUMED") { + eventHandlers.debug?.( + { type: "resumed", data: { shardID: basicShard.id } }, + ); + + basicShard.needToResume = false; + break; + } + // Important for RESUME + if (data.t === "READY") { + basicShard.sessionID = (data.d as ReadyPayload).session_id; + } + + // Update the sequence number if it is present + if (data.s) basicShard.previousSequenceNumber = data.s; + + handleDiscordPayload(data, basicShard.id); + break; } } } -}; +} -function requestGuildMembers( +function identify(shard: BasicShard, payload: IdentifyPayload) { + eventHandlers.debug?.( + { + type: "identifying", + data: { + shardID: shard.id, + }, + }, + ); + + return shard.socket.send( + JSON.stringify( + { + op: GatewayOpcode.Identify, + d: { ...payload, shard: [shard.id, payload.shard[1]] }, + }, + ), + ); +} + +function resume(shard: BasicShard, payload: IdentifyPayload) { + return shard.socket.send(JSON.stringify({ + op: GatewayOpcode.Resume, + d: { + token: payload.token, + session_id: shard.sessionID, + seq: shard.previousSequenceNumber, + }, + })); +} + +async function heartbeat( + shard: BasicShard, + interval: number, + payload: IdentifyPayload, + data: DiscordBotGatewayData, +) { + // We lost socket connection between heartbeats, resume connection + if (shard.socket.isClosed) { + shard.needToResume = true; + resumeConnection(data, payload, shard.id); + heartbeating.delete(shard.id); + return; + } + + if (heartbeating.has(shard.id)) { + const receivedACK = heartbeating.get(shard.id); + // If a ACK response was not received since last heartbeat, issue invalid session close + if (!receivedACK) { + eventHandlers.debug?.( + { + type: "heartbeatStopped", + data: { + interval, + previousSequenceNumber: shard.previousSequenceNumber, + shardID: shard.id, + }, + }, + ); + return shard.socket.send(JSON.stringify({ op: 4009 })); + } + } + + // Set it to false as we are issuing a new heartbeat + heartbeating.set(shard.id, false); + + shard.socket.send( + JSON.stringify( + { op: GatewayOpcode.Heartbeat, d: shard.previousSequenceNumber }, + ), + ); + eventHandlers.debug?.( + { + type: "heartbeat", + data: { + interval, + previousSequenceNumber: shard.previousSequenceNumber, + shardID: shard.id, + }, + }, + ); + await delay(interval); + heartbeat(shard, interval, payload, data); +} + +async function resumeConnection( + data: DiscordBotGatewayData, + payload: IdentifyPayload, + shardID: number, +) { + const shard = basicShards.get(shardID); + if (!shard) { + eventHandlers.debug?.( + { type: "missingShard", data: { shardID: shardID } }, + ); + return; + } + + if (!shard.needToResume) return; + + eventHandlers.debug?.({ type: "resuming", data: { shardID: shard.id } }); + // Run it once + createShard(data, payload, true, shard.id); + // Then retry every 15 seconds + await delay(1000 * 15); + if (shard.needToResume) resumeConnection(data, payload, shardID); +} + +export function requestGuildMembers( guildID: string, + shardID: number, nonce: string, options?: FetchMembersOptions, queuedRequest = false, ) { + const shard = basicShards.get(shardID); + // This request was not from this queue so we add it to queue first if (!queuedRequest) { RequestMembersQueue.push({ guildID, + shardID, nonce, options, }); if (!processQueue) { processQueue = true; - processRequestMembersQueue(); + processGatewayQueue(); } return; } // If its closed add back to queue to redo on resume - if (shardSocket.isClosed) { - requestGuildMembers(guildID, nonce, options); + if (shard?.socket.isClosed) { + requestGuildMembers(guildID, shardID, nonce, options); return; } - shardSocket.send(JSON.stringify({ + shard?.socket.send(JSON.stringify({ op: GatewayOpcode.RequestGuildMembers, d: { guild_id: guildID, query: options?.query || "", - limit: options?.query || 0, + limit: options?.limit || 0, presences: options?.presences || false, user_ids: options?.userIDs, nonce, @@ -231,46 +327,83 @@ function requestGuildMembers( })); } -// TODO: Errors need to be fixed by VSC plugin -// @ts-ignore -postMessage({ type: "REQUEST_CLIENT_OPTIONS" }); -// @ts-ignore -onmessage = (message: MessageEvent) => { - if (message.data.type === "CREATE_SHARD") { - createShard( - message.data.botGatewayData, - message.data.identifyPayload, - ); - shardID = message.data.shardID; +async function processGatewayQueue() { + if (!RequestMembersQueue.length) { + processQueue = false; + return; } - if (message.data.type === "FETCH_MEMBERS") { - requestGuildMembers( - message.data.guildID, - message.data.nonce, - message.data.options, - ); - } + basicShards.forEach((shard) => { + const index = RequestMembersQueue.findIndex((q) => q.shardID === shard.id); + // 2 events per second is the rate limit. + const request = RequestMembersQueue[index]; + if (request) { + eventHandlers.debug?.( + { + type: "requestMembersProcessing", + data: { + remaining: RequestMembersQueue.length, + request, + }, + }, + ); + requestGuildMembers( + request.guildID, + request.shardID, + request.nonce, + request.options, + true, + ); + // Remove item from queue + RequestMembersQueue.splice(index, 1); - if (message.data.type === "EDIT_BOTS_STATUS") { - shardSocket.send(JSON.stringify({ + const secondIndex = RequestMembersQueue.findIndex((q) => + q.shardID === shard.id + ); + const secondRequest = RequestMembersQueue[secondIndex]; + if (secondRequest) { + eventHandlers.debug?.( + { + type: "requestMembersProcessing", + data: { + remaining: RequestMembersQueue.length, + request, + }, + }, + ); + requestGuildMembers( + secondRequest.guildID, + secondRequest.shardID, + secondRequest.nonce, + secondRequest.options, + true, + ); + // Remove item from queue + RequestMembersQueue.splice(secondIndex, 1); + } + } + }); + + await delay(1500); + + processGatewayQueue(); +} + +export function botGatewayStatusRequest(payload: BotStatusRequest) { + basicShards.forEach((shard) => { + shard.socket.send(JSON.stringify({ op: GatewayOpcode.StatusUpdate, d: { since: null, - game: message.data.game.name + game: payload.game.name ? { - name: message.data.game.name, - type: message.data.game.type, + name: payload.game.name, + type: payload.game.type, } : null, - status: message.data.status, + status: payload.status, afk: false, }, })); - } -}; - -function postDebug(details: DebugArg) { - // TODO: Errors need to be fixed by VSC plugin - postMessage({ type: "DEBUG_LOG", details }); + }); } diff --git a/src/module/shardingManager.ts b/src/module/shardingManager.ts index e955bb12e..8895d2a7a 100644 --- a/src/module/shardingManager.ts +++ b/src/module/shardingManager.ts @@ -11,20 +11,11 @@ import { cache } from "../utils/cache.ts"; import { BotStatusRequest } from "../utils/utils.ts"; import { botGatewayStatusRequest, - createBasicShard, + createShard, requestGuildMembers, -} from "./basicShard.ts"; -import { - botGatewayData, - eventHandlers, - IdentifyPayload, - identifyPayload, -} from "./client.ts"; +} from "./shard.ts"; +import { eventHandlers, IdentifyPayload } from "./client.ts"; -let shardCounter = 0; -let basicSharding = false; - -const shards: Worker[] = []; let createNextShard = true; /** This function is meant to be used on the ready event to alert the library to start the next shard. */ @@ -32,58 +23,48 @@ export function allowNextShard(enabled = true) { createNextShard = enabled; } -export function createShardWorker(shardID?: number) { - const path = new URL("./shard.ts", import.meta.url).toString(); - const shard = new Worker(path, { type: "module", deno: true }); - shard.onmessage = (message) => { - if (message.data.type === "REQUEST_CLIENT_OPTIONS") { - identifyPayload.shard = [ - shardID || shardCounter, - botGatewayData.shards, - ]; - - shard.postMessage( - { - type: "CREATE_SHARD", - botGatewayData, - identifyPayload, - shardID: shardCounter, - }, - ); - // Update the shard counter - shardCounter++; - } else if (message.data.type === "HANDLE_DISCORD_PAYLOAD") { - handleDiscordPayload( - JSON.parse(message.data.payload), - message.data.shardID, - ); - } else if (message.data.type === "DEBUG_LOG") { - eventHandlers.debug?.(message.data.details); - } - }; - shards.push(shard); -} - -export const spawnShards = async ( +export async function spawnShards( data: DiscordBotGatewayData, payload: IdentifyPayload, - id = 1, -) => { - if ((data.shards === 1 && id === 1) || id <= data.shards) { - if (createNextShard) { - createNextShard = false; - if (data.shards >= 25) createShardWorker(); - else { - basicSharding = true; - createBasicShard(data, payload, false, id - 1); - } - spawnShards(data, payload, id + 1); - } else { - await delay(1000); - spawnShards(data, payload, id); - } + shardID: number, + lastShardID: number, + skipChecks?: number, +) { + // All shards on this worker have started! Cancel out. + if (shardID >= lastShardID) return; + + if (skipChecks) { + payload.shard = [shardID, data.shards]; + // Start The shard + createShard(data, payload, false, shardID); + // Spawn next shard + spawnShards( + data, + payload, + shardID + 1, + lastShardID, + skipChecks - 1, + ); + return; } -}; + + // Make sure we can create a shard or we are waiting for shards to connect still. + if (createNextShard) { + createNextShard = false; + // Start the next few shards based on max concurrency + spawnShards( + data, + payload, + shardID, + lastShardID, + data.session_start_limit.max_concurrency, + ); + return; + } + + await delay(1000); + spawnShards(data, payload, shardID, lastShardID, skipChecks); +} export async function handleDiscordPayload( data: DiscordPayload, @@ -113,30 +94,13 @@ export async function requestAllMembers( const nonce = `${guild.id}-${Math.random().toString()}`; cache.fetchAllMembersProcessingRequests.set(nonce, resolve); - if (basicSharding) { - return requestGuildMembers(guild.id, guild.shardID, nonce, options); - } - - shards[guild.shardID].postMessage({ - type: "FETCH_MEMBERS", - guildID: guild.id, - nonce, - options, - }); + return requestGuildMembers(guild.id, guild.shardID, nonce, options); } export function sendGatewayCommand(type: "EDIT_BOTS_STATUS", payload: object) { - if (basicSharding) { - if (type === "EDIT_BOTS_STATUS") { - botGatewayStatusRequest(payload as BotStatusRequest); - } - - return; + if (type === "EDIT_BOTS_STATUS") { + botGatewayStatusRequest(payload as BotStatusRequest); } - shards.forEach((shard) => { - shard.postMessage({ - type, - ...payload, - }); - }); + + return; } diff --git a/src/types/discord.ts b/src/types/discord.ts index d750d9e24..d183c8dc0 100644 --- a/src/types/discord.ts +++ b/src/types/discord.ts @@ -57,6 +57,11 @@ export interface DiscordBotGatewayData { remaining: number; /** Milliseconds left until limit is reset. */ reset_after: number; + /** The number of identify requests allowed per 5 seconds. + * So, if you had a max concurrency of 16, and 16 shards for example, you could start them all up at the same time. + * Whereas if you had 32 shards, if you tried to start up shard 0 and 16 at the same time for example, it would not work. You can start shards 0-15 concurrently, then 16-31... + * */ + max_concurrency: number; }; } diff --git a/src/types/options.ts b/src/types/options.ts index 75f6f739d..2a753b587 100644 --- a/src/types/options.ts +++ b/src/types/options.ts @@ -31,6 +31,7 @@ export interface Fulfilled_Client_Options { export interface ClientOptions { token: string; + /** @deprecated Will be removed in next major version! */ properties?: Properties; compress?: boolean; intents: Intents[]; diff --git a/src/utils/constants.ts b/src/utils/constants.ts index ee1ad8a90..5562d5d2d 100644 --- a/src/utils/constants.ts +++ b/src/utils/constants.ts @@ -1,8 +1,15 @@ -export const baseEndpoints = { - BASE_URL: "https://discord.com/api/v8", +// These will never be modified and remain constants +export const discordAPIURLS = { + BASE_URL: `https://discord.com/api/v8`, CDN_URL: "https://cdn.discordapp.com", }; +// This can be modified by big brain bots and use a proxy +export const baseEndpoints = { + BASE_URL: discordAPIURLS.BASE_URL, + CDN_URL: discordAPIURLS.CDN_URL, +}; + const GUILDS_BASE = (id: string) => `${baseEndpoints.BASE_URL}/guilds/${id}`; export const endpoints = {