From be9b78b326a229c9f66c2eb2af356a0e7689fd8f Mon Sep 17 00:00:00 2001 From: Skillz Date: Mon, 26 Oct 2020 11:16:26 -0400 Subject: [PATCH 1/9] start large bot stuff --- src/module/client.ts | 2 +- src/module/hugebot.ts | 20 ++++++++++++++++++++ src/types/options.ts | 1 + 3 files changed, 22 insertions(+), 1 deletion(-) create mode 100644 src/module/hugebot.ts diff --git a/src/module/client.ts b/src/module/client.ts index 53dbcf843..aff72d6cf 100644 --- a/src/module/client.ts +++ b/src/module/client.ts @@ -35,7 +35,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}`; diff --git a/src/module/hugebot.ts b/src/module/hugebot.ts new file mode 100644 index 000000000..8acddedb7 --- /dev/null +++ b/src/module/hugebot.ts @@ -0,0 +1,20 @@ +import { ClientOptions } from "../types/options.ts"; + +/** + * 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 startHugeBot(data: HugeBotOptions) { + +} + +export interface HugeBotOptions extends ClientOptions { + /** This can be used to distribute your bot across different servers. For example, if you wanted 1 million shards per server you could control it using this. */ + shards: [number, 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; +} diff --git a/src/types/options.ts b/src/types/options.ts index 641432ada..1b8d20fa0 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[]; From 9f943cb4a232fae21c8f3bf2dac42c43cb0ab863 Mon Sep 17 00:00:00 2001 From: Skillz Date: Mon, 26 Oct 2020 11:39:09 -0400 Subject: [PATCH 2/9] work --- src/module/hugebot.ts | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/src/module/hugebot.ts b/src/module/hugebot.ts index 8acddedb7..2ce8e0740 100644 --- a/src/module/hugebot.ts +++ b/src/module/hugebot.ts @@ -1,4 +1,12 @@ +import { RequestManager, DiscordBotGatewayData, spawnShards } from "../../mod.ts"; +import { endpoints } from "../constants/discord.ts"; import { ClientOptions } from "../types/options.ts"; +import { botGatewayData, identifyPayload } from "./client.ts"; + +const botOptions = { + token: "", + eventHandlers: {} +} /** * This function should be used only by bot developers whose bots are in over 25,000 servers. @@ -7,7 +15,22 @@ import { ClientOptions } from "../types/options.ts"; * 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 startHugeBot(data: HugeBotOptions) { + botOptions.token = `Bot ${data.token}`; + if (data.eventHandlers) botOptions.eventHandlers = data.eventHandlers; + // Initial API connection to get info about bots connection + botGatewayData = await RequestManager.get( + endpoints.GATEWAY_BOT, + ) as DiscordBotGatewayData; + + identifyPayload.token = data.token; + identifyPayload.intents = data.intents.reduce( + (bits, next) => (bits |= next), + 0, + ); + identifyPayload.shard = [0, botGatewayData.shards]; + + spawnShards(botGatewayData, identifyPayload); } export interface HugeBotOptions extends ClientOptions { From b2424be013452223ed09f8aaa90d41217eb7639e Mon Sep 17 00:00:00 2001 From: Skillz Date: Tue, 27 Oct 2020 07:13:10 -0400 Subject: [PATCH 3/9] more work --- src/module/hugebot.ts | 104 +++++++++++++++++++++++++++++++++--------- src/types/discord.ts | 5 ++ 2 files changed, 88 insertions(+), 21 deletions(-) diff --git a/src/module/hugebot.ts b/src/module/hugebot.ts index 2ce8e0740..d0d30b8f8 100644 --- a/src/module/hugebot.ts +++ b/src/module/hugebot.ts @@ -1,12 +1,27 @@ -import { RequestManager, DiscordBotGatewayData, spawnShards } from "../../mod.ts"; +import { delay } from "../../deps.ts"; +import { DiscordBotGatewayData, RequestManager } from "../../mod.ts"; import { endpoints } from "../constants/discord.ts"; -import { ClientOptions } from "../types/options.ts"; -import { botGatewayData, identifyPayload } from "./client.ts"; +import { ClientOptions, EventHandlers } from "../types/options.ts"; +import { botGatewayData } from "./client.ts"; const botOptions = { - token: "", - eventHandlers: {} -} + workers: new Map(), + eventHandlers: {} as EventHandlers, + botGatewayData: {} as DiscordBotGatewayData, + customShards: [] as number[], + shardsPerWorker: 25, + identifyPayload: { + token: "", + compress: true, + properties: { + $os: "linux", + $browser: "Discordeno", + $device: "Discordeno", + }, + intents: 0, + shard: [0, 0], + }, +}; /** * This function should be used only by bot developers whose bots are in over 25,000 servers. @@ -14,30 +29,77 @@ const botOptions = { * * 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 startHugeBot(data: HugeBotOptions) { - botOptions.token = `Bot ${data.token}`; - if (data.eventHandlers) botOptions.eventHandlers = data.eventHandlers; +export async function startBigBrainBot(data: BigBrainBotOptions) { + botOptions.identifyPayload.token = `Bot ${data.token}`; + if (data.eventHandlers) botOptions.eventHandlers = data.eventHandlers; + if (data.shards) botOptions.customShards = data.shards; + if (data.compress) botOptions.identifyPayload.compress = data.compress; + if (data.shardsPerWorker) botOptions.shardsPerWorker = data.shardsPerWorker; // Initial API connection to get info about bots connection - botGatewayData = await RequestManager.get( + botOptions.botGatewayData = await RequestManager.get( endpoints.GATEWAY_BOT, ) as DiscordBotGatewayData; - identifyPayload.token = data.token; - identifyPayload.intents = data.intents.reduce( + botOptions.identifyPayload.intents = data.intents.reduce( (bits, next) => (bits |= next), 0, ); - identifyPayload.shard = [0, botGatewayData.shards]; + botOptions.identifyPayload.shard = [0, botGatewayData.shards]; - spawnShards(botGatewayData, identifyPayload); + spawnBigBrainBotShards(); } -export interface HugeBotOptions extends ClientOptions { - /** This can be used to distribute your bot across different servers. For example, if you wanted 1 million shards per server you could control it using this. */ - shards: [number, 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; +async function spawnBigBrainBotShards(shardID = 0, skipChecks = 0) { + if (shardID >= botOptions.botGatewayData.shards) return; + + // 25 shards but shards start at 0 so we use 24 + const workerID = shardID % 24; + const worker = botOptions.workers.get(workerID) + + // High max concurrency allows starting shards faster + if (skipChecks) { + // If the worker exists we just need to add + if (worker) { + addShardToWorker(workerID, shardID); + } else { + createShardWorker(workerID, shardID); + } + + spawnBigBrainBotShards(shardID + 1, skipChecks - 1); + } + + // Make sure we can create a shard or we are waiting for shards to connect still. + if (createNextShard) { + // !(shardid % botOptions.botGatewayData.session_start_limit.max_concurrency) + createNextShard = false; + if (botOptions.botGatewayData.shards >= 25) createShardWorker(); + // Start the next few shards based on max concurrency + spawnBigBrainBotShards(shardID + 1, botOptions.botGatewayData.session_start_limit.max_concurrency); + return; + } + + await delay(1000); + spawnBigBrainBotShards(shardID); +} + +export function createShardWorker(workerID: number, shardID: number) { + const path = new URL("./shard.ts", import.meta.url).toString(); + const shard = new Worker(path, { type: "module", deno: true }); + // Add to worker map + botOptions.workers.set(workerID, shard) + +} + +export interface BigBrainBotOptions extends ClientOptions { + /** This can be used to distribute your bot across different servers. For example, if you wanted 1 million shards per server you could control it using this. */ + shards?: [number, 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 allows you to control how many shards per worker. For the times where you can optimize with more shards per worker as your bot has less tasks per shard. + * @default 25 + */ + shardsPerWorker?: number; } diff --git a/src/types/discord.ts b/src/types/discord.ts index 161c15b59..7f48be9ef 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; }; } From bc29fa7d4d40622c1ec409c82a11eb2ee020752c Mon Sep 17 00:00:00 2001 From: Skillz Date: Thu, 29 Oct 2020 20:10:26 -0400 Subject: [PATCH 4/9] big brains --- src/module/bigbrainbot.ts | 105 ++++++++++++++ src/module/bigbrainshard.ts | 272 ++++++++++++++++++++++++++++++++++++ 2 files changed, 377 insertions(+) create mode 100644 src/module/bigbrainbot.ts create mode 100644 src/module/bigbrainshard.ts diff --git a/src/module/bigbrainbot.ts b/src/module/bigbrainbot.ts new file mode 100644 index 000000000..d0d30b8f8 --- /dev/null +++ b/src/module/bigbrainbot.ts @@ -0,0 +1,105 @@ +import { delay } from "../../deps.ts"; +import { DiscordBotGatewayData, RequestManager } from "../../mod.ts"; +import { endpoints } from "../constants/discord.ts"; +import { ClientOptions, EventHandlers } from "../types/options.ts"; +import { botGatewayData } from "./client.ts"; + +const botOptions = { + workers: new Map(), + eventHandlers: {} as EventHandlers, + botGatewayData: {} as DiscordBotGatewayData, + customShards: [] as number[], + shardsPerWorker: 25, + identifyPayload: { + token: "", + compress: true, + properties: { + $os: "linux", + $browser: "Discordeno", + $device: "Discordeno", + }, + intents: 0, + shard: [0, 0], + }, +}; + +/** + * 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) { + botOptions.identifyPayload.token = `Bot ${data.token}`; + if (data.eventHandlers) botOptions.eventHandlers = data.eventHandlers; + if (data.shards) botOptions.customShards = data.shards; + if (data.compress) botOptions.identifyPayload.compress = data.compress; + if (data.shardsPerWorker) botOptions.shardsPerWorker = data.shardsPerWorker; + + // Initial API connection to get info about bots connection + botOptions.botGatewayData = await RequestManager.get( + endpoints.GATEWAY_BOT, + ) as DiscordBotGatewayData; + + botOptions.identifyPayload.intents = data.intents.reduce( + (bits, next) => (bits |= next), + 0, + ); + botOptions.identifyPayload.shard = [0, botGatewayData.shards]; + + spawnBigBrainBotShards(); +} + +async function spawnBigBrainBotShards(shardID = 0, skipChecks = 0) { + if (shardID >= botOptions.botGatewayData.shards) return; + + // 25 shards but shards start at 0 so we use 24 + const workerID = shardID % 24; + const worker = botOptions.workers.get(workerID) + + // High max concurrency allows starting shards faster + if (skipChecks) { + // If the worker exists we just need to add + if (worker) { + addShardToWorker(workerID, shardID); + } else { + createShardWorker(workerID, shardID); + } + + spawnBigBrainBotShards(shardID + 1, skipChecks - 1); + } + + // Make sure we can create a shard or we are waiting for shards to connect still. + if (createNextShard) { + // !(shardid % botOptions.botGatewayData.session_start_limit.max_concurrency) + createNextShard = false; + if (botOptions.botGatewayData.shards >= 25) createShardWorker(); + // Start the next few shards based on max concurrency + spawnBigBrainBotShards(shardID + 1, botOptions.botGatewayData.session_start_limit.max_concurrency); + return; + } + + await delay(1000); + spawnBigBrainBotShards(shardID); +} + +export function createShardWorker(workerID: number, shardID: number) { + const path = new URL("./shard.ts", import.meta.url).toString(); + const shard = new Worker(path, { type: "module", deno: true }); + // Add to worker map + botOptions.workers.set(workerID, shard) + +} + +export interface BigBrainBotOptions extends ClientOptions { + /** This can be used to distribute your bot across different servers. For example, if you wanted 1 million shards per server you could control it using this. */ + shards?: [number, 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 allows you to control how many shards per worker. For the times where you can optimize with more shards per worker as your bot has less tasks per shard. + * @default 25 + */ + shardsPerWorker?: number; +} diff --git a/src/module/bigbrainshard.ts b/src/module/bigbrainshard.ts new file mode 100644 index 000000000..c2e864487 --- /dev/null +++ b/src/module/bigbrainshard.ts @@ -0,0 +1,272 @@ +import type { WebSocket } from "../../deps.ts"; +import { connectWebSocket, delay, isWebSocketCloseEvent } from "../../deps.ts"; +import type { + DiscordBotGatewayData, + DiscordHeartbeatPayload, + ReadyPayload, +} from "../types/discord.ts"; +import { GatewayOpcode } from "../types/discord.ts"; +import type { FetchMembersOptions } from "../types/guild.ts"; +import type { 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; + +const RequestMembersQueue: RequestMemberQueuedRequest[] = []; +let processQueue = false; + +interface RequestMemberQueuedRequest { + guildID: string; + 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, + resuming = false, +) => { + postDebug({ type: "createShard", data: { shardID } }); + + shardSocket = await connectWebSocket(botGatewayData.url); + let resumeInterval = 0; + + if (!resuming) { + // Intial identify with the gateway + await shardSocket.send( + JSON.stringify({ op: GatewayOpcode.Identify, d: identifyPayload }), + ); + } else { + await shardSocket.send(JSON.stringify({ + op: GatewayOpcode.Resume, + d: { + ...identifyPayload, + session_id: sessionID, + seq: previousSequenceNumber, + }, + })); + } + + 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 } }); + + // 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 } }); + + 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)) { + postDebug( + { type: "websocketReconnecting", data: { shardID, message } }, + ); + createShard(botGatewayData, identifyPayload); + } else { + needToResume = true; + resumeConnection(botGatewayData, identifyPayload); + } + } + } +}; + +function requestGuildMembers( + guildID: string, + nonce: string, + options?: FetchMembersOptions, + queuedRequest = false, +) { + // This request was not from this queue so we add it to queue first + if (!queuedRequest) { + RequestMembersQueue.push({ + guildID, + nonce, + options, + }); + + if (!processQueue) { + processQueue = true; + processRequestMembersQueue(); + } + return; + } + + // If its closed add back to queue to redo on resume + if (shardSocket.isClosed) { + requestGuildMembers(guildID, nonce, options); + return; + } + + shardSocket.send(JSON.stringify({ + op: GatewayOpcode.RequestGuildMembers, + d: { + guild_id: guildID, + query: options?.query || "", + limit: options?.query || 0, + presences: options?.presences || false, + user_ids: options?.userIDs, + nonce, + }, + })); +} + +// 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; + } + + if (message.data.type === "FETCH_MEMBERS") { + requestGuildMembers( + message.data.guildID, + message.data.nonce, + message.data.options, + ); + } + + if (message.data.type === "EDIT_BOTS_STATUS") { + shardSocket.send(JSON.stringify({ + op: GatewayOpcode.StatusUpdate, + d: { + since: null, + game: message.data.game.name + ? { + name: message.data.game.name, + type: message.data.game.type, + } + : null, + status: message.data.status, + afk: false, + }, + })); + } +}; + +function postDebug(details: DebugArg) { + // TODO: Errors need to be fixed by VSC plugin + postMessage({ type: "DEBUG_LOG", details }); +} From c2e4c19d4150e2d9092ad565674856b9545e9f10 Mon Sep 17 00:00:00 2001 From: Skillz Date: Fri, 30 Oct 2020 15:47:03 -0400 Subject: [PATCH 5/9] work --- src/module/bigbrainbot.ts | 26 ++++++++++++-------------- 1 file changed, 12 insertions(+), 14 deletions(-) diff --git a/src/module/bigbrainbot.ts b/src/module/bigbrainbot.ts index d0d30b8f8..db30b173a 100644 --- a/src/module/bigbrainbot.ts +++ b/src/module/bigbrainbot.ts @@ -5,6 +5,7 @@ import { ClientOptions, EventHandlers } from "../types/options.ts"; import { botGatewayData } from "./client.ts"; const botOptions = { + createNextShard: false, workers: new Map(), eventHandlers: {} as EventHandlers, botGatewayData: {} as DiscordBotGatewayData, @@ -54,26 +55,29 @@ async function spawnBigBrainBotShards(shardID = 0, skipChecks = 0) { if (shardID >= botOptions.botGatewayData.shards) return; // 25 shards but shards start at 0 so we use 24 - const workerID = shardID % 24; + const workerID = shardID % botOptions.shardsPerWorker - 1; const worker = botOptions.workers.get(workerID) // High max concurrency allows starting shards faster if (skipChecks) { // If the worker exists we just need to add if (worker) { - addShardToWorker(workerID, shardID); + worker.postMessage({ type: "CREATE_SHARD", shardID, workerID, botOptions }); } else { - createShardWorker(workerID, shardID); + const path = new URL("./shard.ts", import.meta.url).toString(); + const newWorker = new Worker(path, { type: "module", deno: true }); + // Add to worker map + botOptions.workers.set(workerID, newWorker); + newWorker.postMessage({ type: "CREATE_SHARD", shardID, workerID, botOptions }); } spawnBigBrainBotShards(shardID + 1, skipChecks - 1); } // Make sure we can create a shard or we are waiting for shards to connect still. - if (createNextShard) { + if (botOptions.createNextShard) { // !(shardid % botOptions.botGatewayData.session_start_limit.max_concurrency) - createNextShard = false; - if (botOptions.botGatewayData.shards >= 25) createShardWorker(); + botOptions.createNextShard = false; // Start the next few shards based on max concurrency spawnBigBrainBotShards(shardID + 1, botOptions.botGatewayData.session_start_limit.max_concurrency); return; @@ -83,14 +87,6 @@ async function spawnBigBrainBotShards(shardID = 0, skipChecks = 0) { spawnBigBrainBotShards(shardID); } -export function createShardWorker(workerID: number, shardID: number) { - const path = new URL("./shard.ts", import.meta.url).toString(); - const shard = new Worker(path, { type: "module", deno: true }); - // Add to worker map - botOptions.workers.set(workerID, shard) - -} - export interface BigBrainBotOptions extends ClientOptions { /** This can be used to distribute your bot across different servers. For example, if you wanted 1 million shards per server you could control it using this. */ shards?: [number, number]; @@ -102,4 +98,6 @@ export interface BigBrainBotOptions extends ClientOptions { * @default 25 */ shardsPerWorker?: number; + /** The absolute file path to the file where the worker will run. */ + workerFilePath?: string; } From 55675937791548e516abe116cc6d77a6e47bdf07 Mon Sep 17 00:00:00 2001 From: Skillz Date: Tue, 3 Nov 2020 15:18:14 -0500 Subject: [PATCH 6/9] cleanup --- src/module/basicShard.ts | 2 +- src/module/bigbrainbot.ts | 59 +++++------------------------------ src/module/shardingManager.ts | 25 +++++++++++++++ 3 files changed, 33 insertions(+), 53 deletions(-) diff --git a/src/module/basicShard.ts b/src/module/basicShard.ts index 60717ca3c..68b9196ff 100644 --- a/src/module/basicShard.ts +++ b/src/module/basicShard.ts @@ -51,7 +51,7 @@ export async function createBasicShard( const basicShard: BasicShard = { id: shardID, - socket: await connectWebSocket(`${data.url}?v=6&encoding=json`), + socket: await connectWebSocket(`${data.url}?v=8&encoding=json`), resumeInterval: 0, sessionID: oldShard?.sessionID || "", previousSequenceNumber: oldShard?.previousSequenceNumber || 0, diff --git a/src/module/bigbrainbot.ts b/src/module/bigbrainbot.ts index db30b173a..aa18c76e7 100644 --- a/src/module/bigbrainbot.ts +++ b/src/module/bigbrainbot.ts @@ -1,16 +1,12 @@ -import { delay } from "../../deps.ts"; -import { DiscordBotGatewayData, RequestManager } from "../../mod.ts"; +import { DiscordBotGatewayData, RequestManager, spawnBigBrainBotShards } from "../../mod.ts"; import { endpoints } from "../constants/discord.ts"; import { ClientOptions, EventHandlers } from "../types/options.ts"; -import { botGatewayData } from "./client.ts"; const botOptions = { createNextShard: false, workers: new Map(), eventHandlers: {} as EventHandlers, botGatewayData: {} as DiscordBotGatewayData, - customShards: [] as number[], - shardsPerWorker: 25, identifyPayload: { token: "", compress: true, @@ -20,7 +16,7 @@ const botOptions = { $device: "Discordeno", }, intents: 0, - shard: [0, 0], + shard: [0, 0] as [number, number], }, }; @@ -33,9 +29,7 @@ const botOptions = { export async function startBigBrainBot(data: BigBrainBotOptions) { botOptions.identifyPayload.token = `Bot ${data.token}`; if (data.eventHandlers) botOptions.eventHandlers = data.eventHandlers; - if (data.shards) botOptions.customShards = data.shards; if (data.compress) botOptions.identifyPayload.compress = data.compress; - if (data.shardsPerWorker) botOptions.shardsPerWorker = data.shardsPerWorker; // Initial API connection to get info about bots connection botOptions.botGatewayData = await RequestManager.get( @@ -46,58 +40,19 @@ export async function startBigBrainBot(data: BigBrainBotOptions) { (bits, next) => (bits |= next), 0, ); - botOptions.identifyPayload.shard = [0, botGatewayData.shards]; - spawnBigBrainBotShards(); + spawnBigBrainBotShards(botOptions.botGatewayData, botOptions.identifyPayload, data.firstShardID, data.lastShardID || (data.firstShardID + 25)); } -async function spawnBigBrainBotShards(shardID = 0, skipChecks = 0) { - if (shardID >= botOptions.botGatewayData.shards) return; - // 25 shards but shards start at 0 so we use 24 - const workerID = shardID % botOptions.shardsPerWorker - 1; - const worker = botOptions.workers.get(workerID) - - // High max concurrency allows starting shards faster - if (skipChecks) { - // If the worker exists we just need to add - if (worker) { - worker.postMessage({ type: "CREATE_SHARD", shardID, workerID, botOptions }); - } else { - const path = new URL("./shard.ts", import.meta.url).toString(); - const newWorker = new Worker(path, { type: "module", deno: true }); - // Add to worker map - botOptions.workers.set(workerID, newWorker); - newWorker.postMessage({ type: "CREATE_SHARD", shardID, workerID, botOptions }); - } - - spawnBigBrainBotShards(shardID + 1, skipChecks - 1); - } - - // Make sure we can create a shard or we are waiting for shards to connect still. - if (botOptions.createNextShard) { - // !(shardid % botOptions.botGatewayData.session_start_limit.max_concurrency) - botOptions.createNextShard = false; - // Start the next few shards based on max concurrency - spawnBigBrainBotShards(shardID + 1, botOptions.botGatewayData.session_start_limit.max_concurrency); - return; - } - - await delay(1000); - spawnBigBrainBotShards(shardID); -} export interface BigBrainBotOptions extends ClientOptions { - /** This can be used to distribute your bot across different servers. For example, if you wanted 1 million shards per server you could control it using this. */ - shards?: [number, number]; + /** 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 allows you to control how many shards per worker. For the times where you can optimize with more shards per worker as your bot has less tasks per shard. - * @default 25 - */ - shardsPerWorker?: number; - /** The absolute file path to the file where the worker will run. */ - workerFilePath?: string; } diff --git a/src/module/shardingManager.ts b/src/module/shardingManager.ts index 4b1cb2d31..68d8e7476 100644 --- a/src/module/shardingManager.ts +++ b/src/module/shardingManager.ts @@ -60,6 +60,31 @@ export function createShardWorker(shardID?: number) { shards.push(shard); } +export async function spawnBigBrainBotShards(data: DiscordBotGatewayData, payload: IdentifyPayload, 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 + createBasicShard(data, payload, false, shardID); + // Spawn next shard + spawnBigBrainBotShards(data, payload, shardID, 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 + spawnBigBrainBotShards(data, payload, shardID + 1, lastShardID, data.session_start_limit.max_concurrency); + return; + } + + await delay(1000); + spawnBigBrainBotShards(data, payload, shardID, lastShardID, skipChecks); +} + export const spawnShards = async ( data: DiscordBotGatewayData, payload: IdentifyPayload, From 46af87f898afef7d1ebb378c40be92eca841e718 Mon Sep 17 00:00:00 2001 From: Skillz Date: Tue, 3 Nov 2020 15:19:22 -0500 Subject: [PATCH 7/9] fixes --- src/module/bigbrainshard.ts | 272 ------------------------------------ src/module/shard.ts | 272 ------------------------------------ 2 files changed, 544 deletions(-) delete mode 100644 src/module/bigbrainshard.ts delete mode 100644 src/module/shard.ts diff --git a/src/module/bigbrainshard.ts b/src/module/bigbrainshard.ts deleted file mode 100644 index c2e864487..000000000 --- a/src/module/bigbrainshard.ts +++ /dev/null @@ -1,272 +0,0 @@ -import type { WebSocket } from "../../deps.ts"; -import { connectWebSocket, delay, isWebSocketCloseEvent } from "../../deps.ts"; -import type { - DiscordBotGatewayData, - DiscordHeartbeatPayload, - ReadyPayload, -} from "../types/discord.ts"; -import { GatewayOpcode } from "../types/discord.ts"; -import type { FetchMembersOptions } from "../types/guild.ts"; -import type { 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; - -const RequestMembersQueue: RequestMemberQueuedRequest[] = []; -let processQueue = false; - -interface RequestMemberQueuedRequest { - guildID: string; - 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, - resuming = false, -) => { - postDebug({ type: "createShard", data: { shardID } }); - - shardSocket = await connectWebSocket(botGatewayData.url); - let resumeInterval = 0; - - if (!resuming) { - // Intial identify with the gateway - await shardSocket.send( - JSON.stringify({ op: GatewayOpcode.Identify, d: identifyPayload }), - ); - } else { - await shardSocket.send(JSON.stringify({ - op: GatewayOpcode.Resume, - d: { - ...identifyPayload, - session_id: sessionID, - seq: previousSequenceNumber, - }, - })); - } - - 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 } }); - - // 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 } }); - - 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)) { - postDebug( - { type: "websocketReconnecting", data: { shardID, message } }, - ); - createShard(botGatewayData, identifyPayload); - } else { - needToResume = true; - resumeConnection(botGatewayData, identifyPayload); - } - } - } -}; - -function requestGuildMembers( - guildID: string, - nonce: string, - options?: FetchMembersOptions, - queuedRequest = false, -) { - // This request was not from this queue so we add it to queue first - if (!queuedRequest) { - RequestMembersQueue.push({ - guildID, - nonce, - options, - }); - - if (!processQueue) { - processQueue = true; - processRequestMembersQueue(); - } - return; - } - - // If its closed add back to queue to redo on resume - if (shardSocket.isClosed) { - requestGuildMembers(guildID, nonce, options); - return; - } - - shardSocket.send(JSON.stringify({ - op: GatewayOpcode.RequestGuildMembers, - d: { - guild_id: guildID, - query: options?.query || "", - limit: options?.query || 0, - presences: options?.presences || false, - user_ids: options?.userIDs, - nonce, - }, - })); -} - -// 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; - } - - if (message.data.type === "FETCH_MEMBERS") { - requestGuildMembers( - message.data.guildID, - message.data.nonce, - message.data.options, - ); - } - - if (message.data.type === "EDIT_BOTS_STATUS") { - shardSocket.send(JSON.stringify({ - op: GatewayOpcode.StatusUpdate, - d: { - since: null, - game: message.data.game.name - ? { - name: message.data.game.name, - type: message.data.game.type, - } - : null, - status: message.data.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/shard.ts b/src/module/shard.ts deleted file mode 100644 index c2e864487..000000000 --- a/src/module/shard.ts +++ /dev/null @@ -1,272 +0,0 @@ -import type { WebSocket } from "../../deps.ts"; -import { connectWebSocket, delay, isWebSocketCloseEvent } from "../../deps.ts"; -import type { - DiscordBotGatewayData, - DiscordHeartbeatPayload, - ReadyPayload, -} from "../types/discord.ts"; -import { GatewayOpcode } from "../types/discord.ts"; -import type { FetchMembersOptions } from "../types/guild.ts"; -import type { 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; - -const RequestMembersQueue: RequestMemberQueuedRequest[] = []; -let processQueue = false; - -interface RequestMemberQueuedRequest { - guildID: string; - 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, - resuming = false, -) => { - postDebug({ type: "createShard", data: { shardID } }); - - shardSocket = await connectWebSocket(botGatewayData.url); - let resumeInterval = 0; - - if (!resuming) { - // Intial identify with the gateway - await shardSocket.send( - JSON.stringify({ op: GatewayOpcode.Identify, d: identifyPayload }), - ); - } else { - await shardSocket.send(JSON.stringify({ - op: GatewayOpcode.Resume, - d: { - ...identifyPayload, - session_id: sessionID, - seq: previousSequenceNumber, - }, - })); - } - - 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 } }); - - // 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 } }); - - 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)) { - postDebug( - { type: "websocketReconnecting", data: { shardID, message } }, - ); - createShard(botGatewayData, identifyPayload); - } else { - needToResume = true; - resumeConnection(botGatewayData, identifyPayload); - } - } - } -}; - -function requestGuildMembers( - guildID: string, - nonce: string, - options?: FetchMembersOptions, - queuedRequest = false, -) { - // This request was not from this queue so we add it to queue first - if (!queuedRequest) { - RequestMembersQueue.push({ - guildID, - nonce, - options, - }); - - if (!processQueue) { - processQueue = true; - processRequestMembersQueue(); - } - return; - } - - // If its closed add back to queue to redo on resume - if (shardSocket.isClosed) { - requestGuildMembers(guildID, nonce, options); - return; - } - - shardSocket.send(JSON.stringify({ - op: GatewayOpcode.RequestGuildMembers, - d: { - guild_id: guildID, - query: options?.query || "", - limit: options?.query || 0, - presences: options?.presences || false, - user_ids: options?.userIDs, - nonce, - }, - })); -} - -// 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; - } - - if (message.data.type === "FETCH_MEMBERS") { - requestGuildMembers( - message.data.guildID, - message.data.nonce, - message.data.options, - ); - } - - if (message.data.type === "EDIT_BOTS_STATUS") { - shardSocket.send(JSON.stringify({ - op: GatewayOpcode.StatusUpdate, - d: { - since: null, - game: message.data.game.name - ? { - name: message.data.game.name, - type: message.data.game.type, - } - : null, - status: message.data.status, - afk: false, - }, - })); - } -}; - -function postDebug(details: DebugArg) { - // TODO: Errors need to be fixed by VSC plugin - postMessage({ type: "DEBUG_LOG", details }); -} From 70ba80b56e7fa4b8ff25b3121ab759d604194db9 Mon Sep 17 00:00:00 2001 From: Skillz Date: Tue, 3 Nov 2020 15:20:10 -0500 Subject: [PATCH 8/9] remove excess --- src/module/hugebot.ts | 105 ------------------------------------------ 1 file changed, 105 deletions(-) delete mode 100644 src/module/hugebot.ts diff --git a/src/module/hugebot.ts b/src/module/hugebot.ts deleted file mode 100644 index d0d30b8f8..000000000 --- a/src/module/hugebot.ts +++ /dev/null @@ -1,105 +0,0 @@ -import { delay } from "../../deps.ts"; -import { DiscordBotGatewayData, RequestManager } from "../../mod.ts"; -import { endpoints } from "../constants/discord.ts"; -import { ClientOptions, EventHandlers } from "../types/options.ts"; -import { botGatewayData } from "./client.ts"; - -const botOptions = { - workers: new Map(), - eventHandlers: {} as EventHandlers, - botGatewayData: {} as DiscordBotGatewayData, - customShards: [] as number[], - shardsPerWorker: 25, - identifyPayload: { - token: "", - compress: true, - properties: { - $os: "linux", - $browser: "Discordeno", - $device: "Discordeno", - }, - intents: 0, - shard: [0, 0], - }, -}; - -/** - * 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) { - botOptions.identifyPayload.token = `Bot ${data.token}`; - if (data.eventHandlers) botOptions.eventHandlers = data.eventHandlers; - if (data.shards) botOptions.customShards = data.shards; - if (data.compress) botOptions.identifyPayload.compress = data.compress; - if (data.shardsPerWorker) botOptions.shardsPerWorker = data.shardsPerWorker; - - // Initial API connection to get info about bots connection - botOptions.botGatewayData = await RequestManager.get( - endpoints.GATEWAY_BOT, - ) as DiscordBotGatewayData; - - botOptions.identifyPayload.intents = data.intents.reduce( - (bits, next) => (bits |= next), - 0, - ); - botOptions.identifyPayload.shard = [0, botGatewayData.shards]; - - spawnBigBrainBotShards(); -} - -async function spawnBigBrainBotShards(shardID = 0, skipChecks = 0) { - if (shardID >= botOptions.botGatewayData.shards) return; - - // 25 shards but shards start at 0 so we use 24 - const workerID = shardID % 24; - const worker = botOptions.workers.get(workerID) - - // High max concurrency allows starting shards faster - if (skipChecks) { - // If the worker exists we just need to add - if (worker) { - addShardToWorker(workerID, shardID); - } else { - createShardWorker(workerID, shardID); - } - - spawnBigBrainBotShards(shardID + 1, skipChecks - 1); - } - - // Make sure we can create a shard or we are waiting for shards to connect still. - if (createNextShard) { - // !(shardid % botOptions.botGatewayData.session_start_limit.max_concurrency) - createNextShard = false; - if (botOptions.botGatewayData.shards >= 25) createShardWorker(); - // Start the next few shards based on max concurrency - spawnBigBrainBotShards(shardID + 1, botOptions.botGatewayData.session_start_limit.max_concurrency); - return; - } - - await delay(1000); - spawnBigBrainBotShards(shardID); -} - -export function createShardWorker(workerID: number, shardID: number) { - const path = new URL("./shard.ts", import.meta.url).toString(); - const shard = new Worker(path, { type: "module", deno: true }); - // Add to worker map - botOptions.workers.set(workerID, shard) - -} - -export interface BigBrainBotOptions extends ClientOptions { - /** This can be used to distribute your bot across different servers. For example, if you wanted 1 million shards per server you could control it using this. */ - shards?: [number, 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 allows you to control how many shards per worker. For the times where you can optimize with more shards per worker as your bot has less tasks per shard. - * @default 25 - */ - shardsPerWorker?: number; -} From af5814e8a7033c04370429f64b8eb13474d355fc Mon Sep 17 00:00:00 2001 From: Skillz Date: Tue, 17 Nov 2020 23:43:04 -0500 Subject: [PATCH 9/9] more work! --- mod.ts | 4 - src/module/bigbrainbot.ts | 58 ------------ src/module/client.ts | 60 +++++++++++- src/module/requestManager.ts | 18 +++- src/module/{basicShard.ts => shard.ts} | 31 +++--- src/module/shardingManager.ts | 125 +++++++------------------ src/utils/constants.ts | 16 ++-- 7 files changed, 131 insertions(+), 181 deletions(-) delete mode 100644 src/module/bigbrainbot.ts rename src/module/{basicShard.ts => shard.ts} (92%) 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/bigbrainbot.ts b/src/module/bigbrainbot.ts deleted file mode 100644 index aa18c76e7..000000000 --- a/src/module/bigbrainbot.ts +++ /dev/null @@ -1,58 +0,0 @@ -import { DiscordBotGatewayData, RequestManager, spawnBigBrainBotShards } from "../../mod.ts"; -import { endpoints } from "../constants/discord.ts"; -import { ClientOptions, EventHandlers } from "../types/options.ts"; - -const botOptions = { - createNextShard: false, - workers: new Map(), - eventHandlers: {} as EventHandlers, - botGatewayData: {} as DiscordBotGatewayData, - identifyPayload: { - token: "", - compress: true, - properties: { - $os: "linux", - $browser: "Discordeno", - $device: "Discordeno", - }, - intents: 0, - shard: [0, 0] as [number, number], - }, -}; - -/** - * 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) { - botOptions.identifyPayload.token = `Bot ${data.token}`; - if (data.eventHandlers) botOptions.eventHandlers = data.eventHandlers; - if (data.compress) botOptions.identifyPayload.compress = data.compress; - - // Initial API connection to get info about bots connection - botOptions.botGatewayData = await RequestManager.get( - endpoints.GATEWAY_BOT, - ) as DiscordBotGatewayData; - - botOptions.identifyPayload.intents = data.intents.reduce( - (bits, next) => (bits |= next), - 0, - ); - - spawnBigBrainBotShards(botOptions.botGatewayData, botOptions.identifyPayload, data.firstShardID, data.lastShardID || (data.firstShardID + 25)); -} - - - -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; -} diff --git a/src/module/client.ts b/src/module/client.ts index f039be0e5..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: "", @@ -51,8 +52,8 @@ export async function createClient(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/basicShard.ts b/src/module/shard.ts similarity index 92% rename from src/module/basicShard.ts rename to src/module/shard.ts index ee1a9cc04..b6ce065cd 100644 --- a/src/module/basicShard.ts +++ b/src/module/shard.ts @@ -7,6 +7,7 @@ import { isWebSocketPongEvent, WebSocket, } from "../../deps.ts"; +import { eventHandlers } from "../../mod.ts"; import { DiscordBotGatewayData, DiscordHeartbeatPayload, @@ -15,7 +16,7 @@ import { } from "../types/discord.ts"; import { FetchMembersOptions } from "../types/guild.ts"; import { BotStatusRequest } from "../utils/utils.ts"; -import { botGatewayData, eventHandlers, IdentifyPayload } from "./client.ts"; +import { IdentifyPayload, proxyWSURL } from "./client.ts"; import { handleDiscordPayload } from "./shardingManager.ts"; const basicShards = new Map(); @@ -40,7 +41,7 @@ interface RequestMemberQueuedRequest { options?: FetchMembersOptions; } -export async function createBasicShard( +export async function createShard( data: DiscordBotGatewayData, identifyPayload: IdentifyPayload, resuming = false, @@ -50,7 +51,9 @@ export async function createBasicShard( const basicShard: BasicShard = { id: shardID, - socket: await connectWebSocket(`${data.url}?v=8&encoding=json`), + socket: await connectWebSocket( + proxyWSURL || `${data.url}?v=8&encoding=json`, + ), resumeInterval: 0, sessionID: oldShard?.sessionID || "", previousSequenceNumber: oldShard?.previousSequenceNumber || 0, @@ -94,10 +97,10 @@ export async function createBasicShard( data: { shardID: basicShard.id, message }, }, ); - createBasicShard(botGatewayData, identifyPayload, false, shardID); + createShard(data, identifyPayload, false, shardID); } else { basicShard.needToResume = true; - resumeConnection(botGatewayData, identifyPayload, basicShard.id); + resumeConnection(data, identifyPayload, basicShard.id); } continue; } else if (isWebSocketPingEvent(message) || isWebSocketPongEvent(message)) { @@ -122,6 +125,7 @@ export async function createBasicShard( basicShard, (data.d as DiscordHeartbeatPayload).heartbeat_interval, identifyPayload, + data, ); } break; @@ -133,7 +137,7 @@ export async function createBasicShard( { type: "reconnect", data: { shardID: basicShard.id } }, ); basicShard.needToResume = true; - resumeConnection(botGatewayData, identifyPayload, basicShard.id); + resumeConnection(data, identifyPayload, basicShard.id); break; case GatewayOpcode.InvalidSession: eventHandlers.debug?.( @@ -141,11 +145,11 @@ export async function createBasicShard( ); // When d is false we need to reidentify if (!data.d) { - createBasicShard(botGatewayData, identifyPayload, false, shardID); + createShard(data, identifyPayload, false, shardID); break; } basicShard.needToResume = true; - resumeConnection(botGatewayData, identifyPayload, basicShard.id); + resumeConnection(data, identifyPayload, basicShard.id); break; default: if (data.t === "RESUMED") { @@ -206,11 +210,12 @@ 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(botGatewayData, payload, shard.id); + resumeConnection(data, payload, shard.id); heartbeating.delete(shard.id); return; } @@ -252,11 +257,11 @@ async function heartbeat( }, ); await delay(interval); - heartbeat(shard, interval, payload); + heartbeat(shard, interval, payload, data); } async function resumeConnection( - botGatewayData: DiscordBotGatewayData, + data: DiscordBotGatewayData, payload: IdentifyPayload, shardID: number, ) { @@ -272,10 +277,10 @@ async function resumeConnection( eventHandlers.debug?.({ type: "resuming", data: { shardID: shard.id } }); // Run it once - createBasicShard(botGatewayData, payload, true, shard.id); + createShard(data, payload, true, shard.id); // Then retry every 15 seconds await delay(1000 * 15); - if (shard.needToResume) resumeConnection(botGatewayData, payload, shardID); + if (shard.needToResume) resumeConnection(data, payload, shardID); } export function requestGuildMembers( diff --git a/src/module/shardingManager.ts b/src/module/shardingManager.ts index 0744efcc4..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,48 +23,28 @@ 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 async function spawnBigBrainBotShards(data: DiscordBotGatewayData, payload: IdentifyPayload, shardID: number, lastShardID: number, skipChecks?: number) { +export async function spawnShards( + data: DiscordBotGatewayData, + payload: IdentifyPayload, + shardID: number, + lastShardID: number, + skipChecks?: number, +) { // All shards on this worker have started! Cancel out. - if (shardID > lastShardID) return; + if (shardID >= lastShardID) return; if (skipChecks) { payload.shard = [shardID, data.shards]; // Start The shard - createBasicShard(data, payload, false, shardID); + createShard(data, payload, false, shardID); // Spawn next shard - spawnBigBrainBotShards(data, payload, shardID, lastShardID, skipChecks - 1); + spawnShards( + data, + payload, + shardID + 1, + lastShardID, + skipChecks - 1, + ); return; } @@ -81,35 +52,20 @@ export async function spawnBigBrainBotShards(data: DiscordBotGatewayData, payloa if (createNextShard) { createNextShard = false; // Start the next few shards based on max concurrency - spawnBigBrainBotShards(data, payload, shardID + 1, lastShardID, data.session_start_limit.max_concurrency); + spawnShards( + data, + payload, + shardID, + lastShardID, + data.session_start_limit.max_concurrency, + ); return; } await delay(1000); - spawnBigBrainBotShards(data, payload, shardID, lastShardID, skipChecks); + spawnShards(data, payload, shardID, lastShardID, skipChecks); } -export const spawnShards = async ( - 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); - } - } -}; - export async function handleDiscordPayload( data: DiscordPayload, shardID: number, @@ -138,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/utils/constants.ts b/src/utils/constants.ts index 3cf526388..5562d5d2d 100644 --- a/src/utils/constants.ts +++ b/src/utils/constants.ts @@ -1,14 +1,14 @@ -let API_VERSION = "v8"; - -export const baseEndpoints = { - /** Although, the version can be defaulted, keep the v6 as it can be changed to test newer versions when necessary. */ - BASE_URL: `https://discord.com/api/${API_VERSION}`, +// These will never be modified and remain constants +export const discordAPIURLS = { + BASE_URL: `https://discord.com/api/v8`, CDN_URL: "https://cdn.discordapp.com", }; -export function changeAPIVersion(number = 7) { - API_VERSION = `v${number}`; -} +// 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}`;